OpenStack 源码系列之 Neutron：Neutron Server 启动

在我们正式开始之前，我们先来看一下博文两张图总结 Neutron 架构总结的 Neutron 架构图（特别简明易懂）。


图片来源

这里我们关注的是 Neutron Server 那一部分的初始化。跟以前 Nova 源码分析一样，我们主要还是会用 UML 来进行代码的分析。

PasteDeploy 特点

这个特点其实在博文 PasteDeploy 其实有说过，就是 PasteDeploy 中初始化 pipeline 和调用 pipeline 的执行顺序是相反的。以 Neutron 的 api-paste.ini 为例：

1
2
3
[composite:neutronapi_v2_0]
...
keystone = ... extensions neutronapiapp_v2_0

初始化该 pipeline(composite) 的顺序是 ...extensions.factory(neutronapiapp_v2_0.factory)，先执行的是（括号里）最后一个应用 neutronapiapp_v2_0 的函数 factory；而调用该 pipeline 就不一样了，执行顺序是 neutronapiapp_v2_0.__call__(extensions.__call__(...))，先执行的也是括号里的函数，最后执行的才是 neutronapiapp_v2_0 的函数 __call__。

注：上边的 __call__ 只是用来表示函数或类调用，实际不一定如此。

Neutron Server 启动概览

其实，Neutron Server 就是一个 WSGI Service，向外接收 REST API 请求，向内路由请求到具体的处理函数。

根据 setup.cfg (neutron-server) 及 api-paste.ini 文件，我们可以得到 Neutron Server 的启动概览：

Neutron API Service 初始化

下文就分为 APIRouter 及 Extensions Routes 两部分进行分析。

在此之前，我们先来看一个 api-paste.ini 文件：

1
2
3
4
5
6
7
8
9
[composite:neutronapi_v2_0]
...
keystone = ... extensions neutronapiapp_v2_0

[filter:extensions]
paste.filter_factory = neutron.api.extensions:plugin_aware_extension_middleware_factory

[app:neutronapiapp_v2_0]
paste.app_factory = neutron.api.v2.router:APIRouter.factory

按照 PasteDeploy 中初始化 pipeline 和调用 pipeline 的执行顺序是相反的的原则，neutronapiapp_v2_0 app 的初始化要早于 extensions app。我们也先从前者（APIRouter）开始说起。

APIRouter

APIRouter 初始化核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class APIRouter(base_wsgi.Router):

    @classmethod
    def factory(cls, global_config, **local_config):
        return cls(**local_config)

    def __init__(self, **local_config):
        mapper = routes_mapper.Mapper()

        # 1.
        # 初始化 NeutronManager，加载了 Core Plugin 及 Service Plugins
        # 而其中 Core Plugin 又会加载 type/mechanism/extension driver
        manager.init()

        plugin = directory.get_plugin()

        # 2.
        # 初始化 PluginAwareExtensionManager (Extension Manager)
        # 实际加载了 neutron/extensions 目录下的所有 extension
        ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
        ext_mgr.extend_resources("2.0", attributes.RESOURCE_ATTRIBUTE_MAP)
        
        # 3.
        # Core plugin 路由映射
        col_kwargs = dict(collection_actions=COLLECTION_ACTIONS,
                          member_actions=MEMBER_ACTIONS)

        def _map_resource(collection, resource, params, parent=None):
            allow_bulk = cfg.CONF.allow_bulk
            controller = base.create_resource(
                collection, resource, plugin, params, allow_bulk=allow_bulk,
                parent=parent, allow_pagination=True,
                allow_sorting=True)
            path_prefix = None
            if parent:
                path_prefix = "/%s/{%s_id}/%s" % (parent['collection_name'],
                                                  parent['member_name'],
                                                  collection)
            mapper_kwargs = dict(controller=controller,
                                 requirements=REQUIREMENTS,
                                 path_prefix=path_prefix,
                                 **col_kwargs)
            return mapper.collection(collection, resource,
                                     **mapper_kwargs)

        mapper.connect('index', '/', controller=Index(RESOURCES))
        for resource in RESOURCES:
            _map_resource(RESOURCES[resource], resource,
                          attributes.RESOURCE_ATTRIBUTE_MAP.get(
                              RESOURCES[resource], dict()))
            resource_registry.register_resource_by_name(resource)

        for resource in SUB_RESOURCES:
            ...

        super(APIRouter, self).__init__(mapper)

代码执行路径可用 UML 表示如下：

NeutronManager

Core Plugin

这里我们所指的 Core Plugin 是 ML2：

1
2
3
# /etc/neutron/neutron.conf
[DEFAULT]
core_plugin = ml2
ML2 简介

博文 OpenStack Neutron ML2 Deep Dive 对 ML2 有一个很简要的介绍：

OpenStack Neutron 作为一种 SDN（Software Defined Network），在其内部使用 ML2 模块来管理 Layer2。ML2 全称是 Modular Layer 2。它是一个可以同时管理多种 Layer 2 技术的框架。在 OpenStack Neutron 的项目代码中，ML2 目前支持 Open vSwitch，linux bridge，SR-IOV 等虚拟化 Layer 2 技术。

它解决了 Neutron 老版本（ML2 之前）存在的两个问题：

  1. 每支持一种 Layer 2 技术，对 OpenStack Neutron 中的 L2 resource（例如 Network/Subnet/Port）的逻辑需要进行一次重写，这大大增加了相应的工作量；
  2. OpenStack Neutron 最多只支持一种 Layer 2 技术，也就是说如果配置使用了 Open vSwitch，那么整个 OpenStack 环境都只能使用 neutron-openvswitch-agent 作为 Layer 2 的管理服务与 Open vSwitch 交互。
    ML2 的提出解决了上面两个问题。ML2 之前的 Layer 2 plugin 代码相同的部分被提取到了 ML2 plugin 中。这样，当一个新的 Layer 2 需要被 Neutron 支持时，只需要实现其特殊部分的代码，需要的代码工作大大减少，开发人员甚至不需要详细了解 Neutron 的具体实现机制，只需要实现对应的接口。

这样，ML2 通过其中的 Mechanism drivers 可以同时管理多种 Layer 2 技术，如下图：

图片来源：OpenStack Neutron ML2 Deep Dive

ML2 的框架如下图所示：

图片来源：OpenStack Neutron ML2 Deep Dive

其中，该博客还对框架中各模块有一个简介（简单汇总了下）：

  • ML2 plugin：所有对 Neutron 中 L2 resource 操作的入口，实现文件是 neutron/plugins/ml2/plugin.py。
  • Extensions drivers：extensions drivers 就是建立 Neutron 中 L2 resource 与其他 resource 之间的联系。当创建、更新或删除 Neutron 中 L2 resource 时，对应的 extensions driver 会被执行，并更新对应的其他 resource。同时 extensions drivers 还会将其他 resource 与 Neutron L2 resource 的关联报告给 ML2 plugin，这样，用户在查看 Port 信息的时候，就能看到对应的 Security Group。
  • Types drivers：物理环境中的 L2 网络类型有很多种，而在虚拟网络 OpenStack Neutron 中，也支持多种网络类型。这些网络类型的支持由 ML2 的 types drivers 来完成。
  • Mechanism drivers：这部分是对各种 L2 技术的支持，例如 Open vSwitch，linux bridge 等等。最近流行的 OVN 也是作为一个 mechanism driver，通过 ML2 与 OpenStack Neutron 工作。
  • RPC：这是 ML2 与 L2 agents 通讯的部分，是基于 AMQP 的实现。例如，删除 Network，需要通过 rpc 通知 L2 agents 也删除相应的流表，虚拟端口等等。
初始化概览

ML2 初始化部分概览如下：

上图中重要的除了一些 driver 的加载，还有一点值得说的是 Neutron 所用的 Callback System。这部分因为在 Service Plugin 也会涉及，所以我们放到本文最后一起说。

DHCP

这里很重要的一点就是初始化 DHCP，导入 network_scheduler_driver：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# neutron/plugins/ml2/plugin.py
class Ml2Plugin(...)

    @resource_registry.tracked_resources(...)
    def __init__(self):
        ...
        self._setup_dhcp()
        ...

    def _setup_dhcp(self):
        """Initialize components to support DHCP."""
        self.network_scheduler = importutils.import_object(
            cfg.CONF.network_scheduler_driver
        )
        self.add_periodic_dhcp_agent_status_check()
RPC Notifiers

在前边初始化分析 Core Plugin 的时候（UML）的时候就已经提到初始化 Core Plugin 的时候就会初始化 RPC Notifiers：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# neutron/plugins/ml2/plugin.py
class Ml2Plugin(...)

    @resource_registry.tracked_resources(...)
    def __init__(self):
        ...
        self._start_rpc_notifiers()
        ...

    @log_helpers.log_method_call
    def _start_rpc_notifiers(self):
        """Initialize RPC notifiers for agents."""
        self.ovo_notifier = ovo_rpc.OVOServerRpcInterface()	# ovo -> oslo versioned object
        self.notifier = rpc.AgentNotifierApi(topics.AGENT)
        self.agent_notifiers[const.AGENT_TYPE_DHCP] = (
            dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
        )

Service Plugins

Service Plugins 比较多，如我们在 setup.cfg 文件就可以看到声明的诸多 plugin：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# setup.cfg
neutron.service_plugins =
    dummy = neutron.tests.unit.dummy_plugin:DummyServicePlugin
    router = neutron.services.l3_router.l3_router_plugin:L3RouterPlugin
    metering = neutron.services.metering.metering_plugin:MeteringPlugin
    qos = neutron.services.qos.qos_plugin:QoSPlugin
    tag = neutron.services.tag.tag_plugin:TagPlugin
    flavors = neutron.services.flavors.flavors_plugin:FlavorsPlugin
    auto_allocate = neutron.services.auto_allocate.plugin:Plugin
    segments = neutron.services.segments.plugin:Plugin
    network_ip_availability = neutron.services.network_ip_availability.plugin:NetworkIPAvailabilityPlugin
    revisions = neutron.services.revisions.revision_plugin:RevisionPlugin
    timestamp = neutron.services.timestamp.timestamp_plugin:TimeStampPlugin
    trunk = neutron.services.trunk.plugin:TrunkPlugin
    loki = neutron.services.loki.loki_plugin:LokiPlugin

但实际只加载 cfg.CONF.service_plugins 及 DEFAULT_SERVICE_PLUGINS 所限定的。

下边我们以 L3RouterPlugin 为例来做一个简要说明。

L3 Router Plugin

一个粗略的加载图如下：

Service Plugin 的详细加载就不细说了，我们会在后续博文中单位为一些重要的 plguin 做分析，那时再做详细解析。

RPC Notifiers

如 L3 Router Plugin 定义的 RPC Notifiers 如下：

1
2
3
4
5
6
7
class L3RouterPlugin(...):
    ...
    @resource_registry.tracked_resources(...)
    def __init__(self):
        ...
        self.agent_notifiers.update(
            {n_const.AGENT_TYPE_L3: l3_rpc_agent_api.L3AgentNotifyAPI()})

Extension Manager

在 APIRouter 初始化完 Core Plugin 和 Service Plugins 之后就是加载 PluginAwareExtensionManager(Extension Manager)，进一步又会把 neutron/extensions 目录下的 Extension(API) 都加载，具体如下：

Core Plugin Routes

加载完 Core Plugin 及 Service Plugins 之后，接下来一步就是建立路由映射了，不过这里的路由映射只做了 Core Plugin 的路由映射（network/subnet/subnetpool/port）。具体看代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
# neutron/api/v2/router.py
RESOURCES = {'network': 'networks',
             'subnet': 'subnets',
             'subnetpool': 'subnetpools',
             'port': 'ports'}	# !!!important
SUB_RESOURCES = {}

class APIRouter(base_wsgi.Router):
    def __init__(self, **local_config):
        mapper = routes_mapper.Mapper()
        ...
        plugin = directory.get_plugin()	# get the core plugin instance
        ...

        col_kwargs = dict(collection_actions=COLLECTION_ACTIONS,
                          member_actions=MEMBER_ACTIONS)

        def _map_resource(collection, resource, params, parent=None):
            allow_bulk = cfg.CONF.allow_bulk
            # collection 为资源复数形式，如 networks; resource 为资源单数形式，如 network。
            controller = base.create_resource(
                collection, resource, plugin, params, allow_bulk=allow_bulk,
                parent=parent, allow_pagination=True,
                allow_sorting=True)
            path_prefix = None
            if parent:
                path_prefix = "/%s/{%s_id}/%s" % (parent['collection_name'],
                                                  parent['member_name'],
                                                  collection)
            mapper_kwargs = dict(controller=controller,
                                 requirements=REQUIREMENTS,
                                 path_prefix=path_prefix,
                                 **col_kwargs)
            return mapper.collection(collection, resource,
                                     **mapper_kwargs)

        mapper.connect('index', '/', controller=Index(RESOURCES))
        for resource in RESOURCES:	# !!!important
            _map_resource(RESOURCES[resource], resource,
                          attributes.RESOURCE_ATTRIBUTE_MAP.get(
                              RESOURCES[resource], dict()))
            resource_registry.register_resource_by_name(resource)

        for resource in SUB_RESOURCES:
            ...

关于路由映射可参考之前博文 routes。不过这里想重点说一下 create_resource 这个函数，它返回的是一个 (WSGI) Controller。这个函数很重要，Core Plugin、Extensions 路由映射都会调用它。下边我们来重点看下：

1
2
3
4
5
6
7
8
9
10
# neutron/api/v2/base.py
def create_resource(collection, resource, plugin, params, allow_bulk=False,
                    member_actions=None, parent=None, allow_pagination=False,
                    allow_sorting=False):
    controller = Controller(plugin, collection, resource, params, allow_bulk,
                            member_actions=member_actions, parent=parent,
                            allow_pagination=allow_pagination,
                            allow_sorting=allow_sorting)

    return wsgi_resource.Resource(controller, FAULT_MAP)

如果我们看一下 Controller 这个类，我们很容易就可以看出它究竟做的是什么事情了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
# neutron/api/v2/base.py
class Controller(object):
    LIST = 'list'
    SHOW = 'show'
    CREATE = 'create'
    UPDATE = 'update'
    DELETE = 'delete'

    def __init__(self, plugin, collection, resource, attr_info,
                 allow_bulk=False, member_actions=None, parent=None,
                 allow_pagination=False, allow_sorting=False):
        if member_actions is None:
            member_actions = []
        self._plugin = plugin	# 实际执行者，Core Plugin 或 Service Plugin
        self._collection = collection.replace('-', '_')
        self._resource = resource.replace('-', '_')
        ...
        self._member_actions = member_actions
        ...
        self.parent = parent
        if parent:
            self._parent_id_name = '%s_id' % parent['member_name']
            parent_part = '_%s' % parent['member_name']
        else:
            self._parent_id_name = None
            parent_part = ''
        self._plugin_handlers = {
            self.LIST: 'get%s_%s' % (parent_part, self._collection),
            self.SHOW: 'get%s_%s' % (parent_part, self._resource)
        }
        for action in [self.CREATE, self.UPDATE, self.DELETE]:
            self._plugin_handlers[action] = '%s%s_%s' % (action, parent_part,
                                                         self._resource)

    @db_api.retry_db_errors
    def index(self, request, **kwargs):
        """Returns a list of the requested entity."""
        ...

    @db_api.retry_db_errors
    def show(self, request, id, **kwargs):
        """Returns detailed information about the requested entity."""
        ...

    def create(self, request, body=None, **kwargs):
        ...
        return self._create(request, body, **kwargs)

    @db_api.retry_db_errors
    def _create(self, request, body, **kwargs):
        """Creates a new instance of the requested entity."""
        ...

    def update(self, request, id, body=None, **kwargs):
        """Updates the specified entity's attributes."""
        ...

    def delete(self, request, id, **kwargs):
        """Deletes the specified entity."""
        ...

也就是说，它定义了一些资源的常见操作（CRUD），然后再根据具体资源及 action 将这些操作映射到对 plugin (Core Plguin or Service Plugin) 的真正操作上，如 show 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# neutron/api/v2/base.py
class Controller(object):
    @db_api.retry_db_errors
    def show(self, request, id, **kwargs):
        """Returns detailed information about the requested entity."""
        try:
            ...
            return {..., self._item(...)}

    def _item(self, request, id, do_authz=False, field_list=None,
              parent_id=None):
        """Retrieves and formats a single element of the requested entity."""
        kwargs = {'fields': field_list}
        action = self._plugin_handlers[self.SHOW]
        if parent_id:
            kwargs[self._parent_id_name] = parent_id
        obj_getter = getattr(self._plugin, action)
        obj = obj_getter(request.context, id, **kwargs)
        ...
        return obj

这样确实能够统一处理所有 plugin 的路由映射了，只要它们的 action 符合这个 Controller 定义的标准即可。但是也正因为路由构造如此高度抽象，一开始看很容易看晕…

不过，这个 Controller 还不是一个 WSGI 应用，需要继续进一步包装：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# neutron/api/v2/base.py
def create_resource(collection, resource, plugin, params, allow_bulk=False,
                    member_actions=None, parent=None, allow_pagination=False,
                    allow_sorting=False):
    ...
    return wsgi_resource.Resource(controller, FAULT_MAP)

# neutron/api/v2/resource.py
def Resource(controller, ...):
    ...
    @webob.dec.wsgify(RequestClass=Request)
    def resource(request):	# 类似于下例中 BaseController 的 __call__ 函数
        route_args = request.environ.get('wsgiorg.routing_args')
        if route_args:
            args = route_args[1].copy()
        else:
            args = {}

        ...
        action = args.pop('action', None)
        ...

        try:
            ...
            method = getattr(controller, action)
            result = method(request=request, **args)
        except Exception as e:
            ...

        return webob.Response(request=request, status=status,
                              content_type=content_type,
                              body=body)
    ...
    return resource

一看就很熟悉对不对？我们只要比较一下 OpenStack 其他模块（如 Nova）常用的路由 Controller 构造方法就很清楚了（样例）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import webob.dec
import webob.exc
import routes.middleware
from wsgiref.simple_server import make_server


# Environment variable used to pass the request context
CONTEXT_ENV = 'openstack.context'

# Environment variable used to pass the request params
PARAMS_ENV = 'openstack.params'


class BaseController(object):
    def _normalize_arg(self, arg):
        return arg.replace(':', '_').replace('-', '_')

    def _normalize_dict(self, d):
        return {self._normalize_arg(k): v for (k, v) in d.items()}

    def _render_response(self, body=None, status=None, method=None):
        '''Form a WSGI response.'''
        resp = webob.Response(body=body, status=status)
        resp.method = method
        return resp

    @webob.dec.wsgify
    def __call__(self, req):
        arg_dict = req.environ['wsgiorg.routing_args'][1]
        action = arg_dict.pop('action')
        del arg_dict['controller']

        params = req.environ.get(PARAMS_ENV, {})
        params.update(arg_dict)

        method = getattr(self, action)
        params = self._normalize_dict(params)
        result = method(req, **params)

        return self._render_response(body=result.body, status=result.status, method=req.method)


class UserController(BaseController):
    def get_user(self, req, user_id):
        return webob.Response('I am a user, my ID is %s.' % user_id)


class TenantController(BaseController):
    def get_tenant(self, req, tenant_id):
        return webob.Response('I am a tenant, my ID is %s.' % tenant_id)


class Router(object):
    def __init__(self):
        self._mapper = routes.Mapper()
        self._mapper.connect('/users/{user_id}',
                             controller=UserController(),
                             action='get_user',
                             conditions={'method': ['GET']})
        self._mapper.connect('/tenants/{tenant_id}',
                             controller=TenantController(),
                             action='get_tenant',
                             conditions={'method': ['GET']})

        self._router = routes.middleware.RoutesMiddleware(self._dispatch, self._mapper)

    @webob.dec.wsgify
    def __call__(self, req):
        return self._router

    @staticmethod
    @webob.dec.wsgify
    def _dispatch(req):
        match = req.environ['wsgiorg.routing_args'][1]

        if not match:
            return webob.exc.HTTPNotFound()

        app = match['controller']
        return app


router = Router()
httpd = make_server('localhost', 8080, router)
httpd.serve_forever()

** Extensions Routes**

虽然前边 Service Plugins 已经加载完毕了，但它们的路由还没建立…

本文一开始就在提到 PasteDeploy 中初始化 pipeline 和调用 pipeline 的执行顺序是相反的，这也就是为什么在 api-paste.ini 中 extensions app 的初始化要在 neutronapiapp_v2_0 app 之后。且看详细：：

1
2
3
4
5
6
[composite:neutronapi_v2_0]
...
keystone = ... extensions neutronapiapp_v2_0

[filter:extensions]
paste.filter_factory = neutron.api.extensions:plugin_aware_extension_middleware_factory

下边就直接以代码来说明。

1
2
3
4
5
6
7
# neutron/api/extensions.py
def plugin_aware_extension_middleware_factory(global_config, **local_config):
    """Paste factory."""
    def _factory(app):
        ext_mgr = PluginAwareExtensionManager.get_instance()	# 这一步已经在 APIRouter 中初始化了，直接返回
        return ExtensionMiddleware(app, ext_mgr=ext_mgr)
    return _factory

也就是说，这里分成了两步，第 1 步是加载扩展 Extension Manager，第 2 步才是做路由映射。第 1 步已经在 APIRouter 中完成了，接下来我们要看的是第 2 步。

直接上代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# neutron/api/extensions.py
class ExtensionMiddleware(base.ConfigurableMiddleware):
    """Extensions middleware for WSGI."""

    def __init__(self, application,
                 ext_mgr=None):
        self.ext_mgr = (ext_mgr or ExtensionManager(get_extensions_path()))	# 在 APIRouter 已初始化过
        mapper = routes.Mapper()

        # extended resources
        for resource in self.ext_mgr.get_resources():	# !!! get_resources 函数是重点
            ...

            # !!! 重点
            mapper.resource(resource.collection, resource.collection,
                            controller=resource.controller,
                            member=resource.member_actions,
                            parent_resource=resource.parent,
                            path_prefix=path_prefix)

        ...
        self._router = routes.middleware.RoutesMiddleware(self._dispatch, mapper)
        super(ExtensionMiddleware, self).__init__(application)

self.ext_mgr.get_resources() 实际调用的是每个 extension 的 get_resources 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# neutron/api/extensions.py
class ExtensionMiddleware(base.ConfigurableMiddleware):
    """Extensions middleware for WSGI."""

    def get_resources(self):
        """Returns a list of ResourceExtension objects."""
        resources = []
        # 添加一个 extension 本身的 CRUD，如 list extensions 可以拿到
        # neutron 所加载的 extension 列表，命令行为 neutron ext-list
        resources.append(ResourceExtension('extensions', ExtensionController(self)))
        # 一个个调用 extension 的 get_resources 函数
        for ext in self.extensions.values():
            resources.extend(ext.get_resources())
        return resources

下边我们以 L3 extension 为例来说明实际的路由映射是怎么做的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# neutron/extensions/l3.py
class L3(extensions.ExtensionDescriptor):
    @classmethod
    def get_resources(cls):
        """Returns Ext Resources."""
        plural_mappings = resource_helper.build_plural_mappings(
            {}, RESOURCE_ATTRIBUTE_MAP)
        # member actions
        action_map = {'router': {'add_router_interface': 'PUT',
                                 'remove_router_interface': 'PUT'}}
        return resource_helper.build_resource_info(plural_mappings,
                                                   RESOURCE_ATTRIBUTE_MAP,
                                                   constants.L3,
                                                   action_map=action_map,
                                                   register_quota=True)
# neutron/api/v2/resource_helper.py
def build_resource_info(plural_mappings, resource_map, which_service,
                        action_map=None, register_quota=False,
                        translate_name=False, allow_bulk=False):
    """Build resources for advanced services.
       ...
    """
    resources = []
    if not which_service:
        which_service = constants.CORE
    if action_map is None:
        action_map = {}
    # !!! 重要
    plugin = directory.get_plugin(which_service)
    path_prefix = getattr(plugin, "path_prefix", "")
    LOG.debug('Service %(service)s assigned prefix: %(prefix)s',
              {'service': which_service, 'prefix': path_prefix})
    for collection_name in resource_map:
        resource_name = plural_mappings[collection_name]
        params = resource_map.get(collection_name, {})
        if translate_name:
            collection_name = collection_name.replace('_', '-')
        if register_quota:
            resource_registry.register_resource_by_name(resource_name)
        member_actions = action_map.get(resource_name, {})
        # !!! 重要
        controller = base.create_resource(
            collection_name, resource_name, plugin, params,
            member_actions=member_actions,
            allow_bulk=allow_bulk,
            allow_pagination=True,
            allow_sorting=True)
        resource = extensions.ResourceExtension(
            collection_name,
            controller,
            path_prefix=path_prefix,
            member_actions=member_actions,
            attr_map=params)
        resources.append(resource)
    return resources

上边有两个地方很重：

  1. 获取 plugin，这里具体获取的是 neutron.services.l3_router.l3_router_plugin.L3RouterPlugin，它的初始化在前文说过了；
  2. extension 的路由映射和 Core Plugin 一样最终都会调用 create_resource 这个函数。接下来的步骤参考 Core Plugin 的就行了。

注：如果我们仔细看，有些 Extension（如 timestamp） 的 get_resources 返回的是一个空列表，也就是说，这些 Extension 不会做路由映射。

Extension 与 Service Plugin 关系

对于 Core Plugin 来说，它的路由映射是单独做的，不容易误解；但对 Service Plugin 来说，它的路由是由 Extension 来做的。把 Extension 成为 Extension API 更为合适。

API 与 Plugin 之间的关系可参考陈沙克老师画的一张图：

服务启动

这里的服务指的是 NeutronAPIServie、RPC 以及 Plugins（Core Plugin & Service Plugin）。服务启动部分我们在 "Neutron Server 启动概览" 部分的 UML 图中均有体现出来（_run_wsgi 及 start_api_and_rpc_workers 函数）。涉及代码如下：

1
2
3
4
# neutron/server/wsgi_eventlet.py
def eventlet_wsgi_server():
    neutron_api = service.serve_wsgi(service.NeutronApiService)
    start_api_and_rpc_workers(neutron_api)

NeutronApiService

NeutronApiService 的启动在 _run_wsgi 函数中：

1
2
3
4
5
6
# neutron/service.py
def _run_wsgi(app_name):
    app = config.load_paste_app(app_name)	# 初始化 NeutronApiService 前文已经分析过
    ...
    return run_wsgi_app(app)

它做的第 1 件事初始化 NeutronApiService 前文已经分析过，接下来我们要来看一下它如何启动 NeutronApiService：

下边我们就以启动 NeutronApiService 为例来说明启动详细过程（不包括 evenlet 部分）：

Plugins RPC Listeners & Workers

这里的 Plugins 包括了 Core Plugin 及 Service Plugins。它们的 RPC Listeners & Workers 启动在 start_api_and_rpc_workers 的函数中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# neutron/server/wsgi_eventlet.py
def eventlet_wsgi_server():
    ...
    start_api_and_rpc_workers(neutron_api)

def start_api_and_rpc_workers(neutron_api):
    try:
        worker_launcher = service.start_all_workers()
        ...

# neutron/service.py
def start_all_workers():
    workers = _get_rpc_workers() + _get_plugins_workers()
    return _start_workers(workers)

详细分析见如下的 UML 图：

RPC Listeners(Workers)

这里我们以 Core Plugin 为例（）。Core Plugin 的 start_rpc_listeners、start_rpc_state_reports_listener (只有 Core Plugin 才有) 定义如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# neutron/plugins/ml2/plugin.py
class Ml2Plugin(...):

    @log_helpers.log_method_call
    def start_rpc_listeners(self):
        """Start the RPC loop to let the plugin communicate with agents."""
        self._setup_rpc()
        self.topic = topics.PLUGIN
        self.conn = n_rpc.create_connection()
        self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
        self.conn.create_consumer(
            topics.SERVER_RESOURCE_VERSIONS,
            [resources_rpc.ResourcesPushToServerRpcCallback()],
            fanout=True)
        # process state reports despite dedicated rpc workers
        self.conn.create_consumer(topics.REPORTS,
                                  [agents_db.AgentExtRpcCallback()],
                                  fanout=False)
        return self.conn.consume_in_threads()

    def start_rpc_state_reports_listener(self):
        self.conn_reports = n_rpc.create_connection()
        self.conn_reports.create_consumer(topics.REPORTS,
                                          [agents_db.AgentExtRpcCallback()],
                                          fanout=False)
        return self.conn_reports.consume_in_threads()

Plugin Workers

而关于 Core & Service Plugins 的 Workers 的获取则是调用每个 Plugin 的 get_workers 函数，具体代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
# neutron/service.py
def _get_plugins_workers():
    # NOTE(twilson) get_plugins also returns the core plugin
    plugins = directory.get_unique_plugins()

    # TODO(twilson) Instead of defaulting here, come up with a good way to
    # share a common get_workers default between NeutronPluginBaseV2 and
    # ServicePluginBase
    return [
        plugin_worker
        for plugin in plugins if hasattr(plugin, 'get_workers')
        for plugin_worker in plugin.get_workers()
    ]

启动 RPC & Plugins Workers

获取到 RPC & Plugins Workers 之后就是要启动这些 Workers，具体代码在 _start_workers 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# neutron/service.py
def _start_workers(workers):
    process_workers = [
        plugin_worker for plugin_worker in workers
        if plugin_worker.worker_process_count > 0
    ]

    try:
        if process_workers:
            worker_launcher = common_service.ProcessLauncher(
                cfg.CONF, wait_interval=1.0
            )
            ...
            for worker in process_workers:
                worker_launcher.launch_service(worker,
                                               worker.worker_process_count)
        ...

这个启动方式我们在 NeutronAPIService 服务启动里分析过，请参考之。

Wait for NeutronApiService/RPC/Plugins Workers

当 NeutronApiService/RPC/Plugins Workers (Services) 都启动起来后，就会用 eventlet thread 来调用这些 Workers (Services) 的 wait 函数（如 oslo_service.service::ProcessLauncher、neutron.wsgi::WorkerService 的 wait 函数）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# neutron/server/wsgi_eventlet.py
def start_api_and_rpc_workers(neutron_api):
    try:
        worker_launcher = service.start_all_workers()

        pool = eventlet.GreenPool()
        api_thread = pool.spawn(neutron_api.wait)
        plugin_workers_thread = pool.spawn(worker_launcher.wait)

        # api and other workers should die together. When one dies,
        # kill the other.
        api_thread.link(lambda gt: plugin_workers_thread.kill())
        plugin_workers_thread.link(lambda gt: api_thread.kill())

        pool.waitall()
    ...

附：Callback System

简介

不同于 Nova，Neutron 多了回调系统（Callback System）和消息回调系统（Messaging Callback System）。这两种系统的区别在于用途及实现方式：

Neutron already has a callback system for in-process resource callbacks where publishers and subscribers are able to publish and subscribe for resource events.
This system (Messaging Callback System) is different, and is intended to be used for inter-process callbacks, via the messaging fanout mechanisms.

也就是说，回调系统用于进程内通信（如 Neutron Server 进程里的 core plugin 与 service plugin 通信）；消息回调系统用于进程间通信，通过消息队列的广播机制实现。前者相对于后者来说比较简单，无需通过复杂机制来实现。

这里其实还有一个疑问，就是既然是进程内通信了，那还有必要弄一个回调系统吗？应该说，不用该系统也是可以实现进程内各服务之间的通信的，但考虑到 Neutron 的复杂性（多 service plugin），不用回调系统会使得各服务之间强耦合，而且不便于 service 独立开发：

In Neutron, core and service components may need to cooperate during the execution of certain operations, or they may need to react upon the occurrence of certain events. For instance, when a Neutron resource is associated to multiple services, the components in charge of these services may need to play an active role in determining what the right state of the resource needs to be.
The cooperation may be achieved by making each object aware of each other, but this leads to tight coupling, or alternatively it can be achieved by using a callback-based system, where the same objects are allowed to cooperate in a loose manner.
This is particularly important since the spin off of the advanced services like VPN, Firewall and Load Balancer, where each service’s codebase lives independently from the core and from one another. This means that the tight coupling is no longer a practical solution for object cooperation. In addition to this, if more services are developed independently, there is no viable integration between them and the Neutron core. A callback system, and its registry, tries to address these issues.

这样使用回调系统的好处就是解耦，各个 service 只需知道作为通信的中介（intermediary）即可：

The intermediary is then in charge of keeping track of who is interested in the messages and in delivering the messages forth and back, when required. As mentioned earlier, the use of an intermediary has the benefit of decoupling the parties involved in the communications, as now they only need to know about the intermediary; the other benefit is that the use of an intermediary opens up the possibility of multiple party communication: more than one object can express interest in receiving the same message, and the same message can be delivered to more than one object.

对于一个回调系统而言，我们需要考虑以下三个问题：

  • how to become consumer of messages (i.e. how to be on the receiving end of the message);
  • how to become producer of messages (i.e. how to be on the sending end of the message);
  • how to consume/produce messages selectively;

针对 Neutron 而言，该系统主要用于资源的生命周期事件方面：

Translate and narrow this down to Neutron needs, and this means the design of a callback system where messages are about lifecycle events (e.g. before creation, before deletion, etc.) of Neutron resources (e.g. networks, routers, ports, etc.).

源码解析

回调系统的代码还挺简单，很容易看懂。代码路径：neutron.callbacks.manager.py::CallbacksManager。

subscribe & unsubscribe

订阅与取消订阅实际就是注册或注销回调函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# neutron.callbacks.manager.py
class CallbacksManager(object):
    """A callback system that allows objects to cooperate in a loose manner."""

    ...
    def subscribe(self, callback, resource, event):
        """Subscribe callback for a resource event.

        The same callback may register for more than one event.

        :param callback: the callback. It must raise or return a boolean.
        :param resource: the resource. It must be a valid resource.
        :param event: the event. It must be a valid event.
        """
        LOG.debug("Subscribe: %(callback)s %(resource)s %(event)s",
                  {'callback': callback, 'resource': resource, 'event': event})

        callback_id = _get_id(callback)
        try:
            self._callbacks[resource][event][callback_id] = callback
        ...
        self._index[callback_id][resource].add(event)

    def unsubscribe(self, callback, resource, event):
        """Unsubscribe callback from the registry.

        :param callback: the callback.
        :param resource: the resource.
        :param event: the event.
        """
        ...
        callback_id = self._find(callback)
        if not callback_id:
            LOG.debug("Callback %s not found", callback_id)
            return
        if resource and event:
            del self._callbacks[resource][event][callback_id]
            self._index[callback_id][resource].discard(event)
            ...

notify

通知也挺简单的，就是遍历，然后一个一个调用订阅某一事件的回调函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# neutron.callbacks.manager.py
class CallbacksManager(object):
    ...

    def notify(self, resource, event, trigger, **kwargs):
        """Notify all subscribed callback(s).

        Dispatch the resource's event to the subscribed callbacks.

        :param resource: the resource.
        :param event: the event.
        :param trigger: the trigger. A reference to the sender of the event.
        """
        errors = self._notify_loop(resource, event, trigger, **kwargs)
        ...

    def _notify_loop(self, resource, event, trigger, **kwargs):
        """The notification loop."""
        errors = []
        callbacks = list(self._callbacks[resource].get(event, {}).items())
        LOG.debug("Notify callbacks %s for %s, %s",
                  callbacks, resource, event)
        # TODO(armax): consider using a GreenPile
        for callback_id, callback in callbacks:
            try:
                callback(resource, event, trigger, **kwargs)
            ...

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

谢宏峰，毕业于暨南大学（2009 - 2013）、中国科学院（2013 - 2016），现就职于深信服科技，从事 OpenStack 开发。