【neutron源码分析】neutron-server启动流程分析

拼搏现实的明天。 2022-06-12 06:29 575阅读 0赞

Neutron为Openstack的网络组件，其内部功能均是以plugin形式实现的，其中代表性的plugin就是ml2和l3，下面将从neutron启动的源码来分析neutron加载和扩展插件的流程。

## 1 概念 ##

** (1) TAP/TUN/VETCH**

**  ** TAP/TUN是Linux内核实现的一对虚拟网络设备，***TAP工作在二层，TUN工作在三层。***

TAP可以实现虚拟网卡的功能，***虚拟机的每个vNIC都与Hypervisor中的一个TAP设备相连***。当一个TAP设备被创建时，在Linux设备文件目录下将会生成一个对应的字符设备文件，用户程序可以像打开普通文件一样打开这个文件进行读写。

当对这个TAP设备文件执行write操作时，对于Linux网络子系统来说，就相当于TAP设备收到了数据，并请求内核接受它，Linux内核收到此数据后将根据网络配置进行后续处理，处理过程类似于普通的物理网卡从外界接收数据。当用户程序执行read请求时，相当于向内核查询TAP设备上是否有数据要被发送，有的话则取出到用户程序里，从而完成TAP设备发送数据的功能。在这个过程里，TAP当一个网卡，而操作TAP设备的应用程序相当于另外一台计算机，它通过read/write系统调用，和本机进行网络通信，Subnet属于网路中的3层，IPv4或IPv6地址并描述其相关的配置信息，它附加在一个二层network上并指明属于这个network的虚拟机可使用的IP地址范围。

VETH设备总是成对出现，送到一端请求发送的数据总是从另一端以请求接受的形式出现。创建并配置正确后，向其一端输入数据，VETH会改变数据的方向并将其送入内核网络子系统，完成数据的注入，而在另一端则能读到此数据。

--------------------

## 2 启动 ##

**命令行：**

neutron-server --config-file /usr/share/neutron/neutron-dist.conf --config-dir /usr/share/neutron/server --config-file /etc/neutron/neutron.conf --config-file /etc/neutron/plugin.ini --config-dir /etc/neutron/conf.d/common --config-dir /etc/neutron/conf.d/neutron-server --log-file /var/log/neutron/server.log

#### setup.cfg--entrypoint 配置 ####

[entry_points]
    console_scripts =
        neutron-db-manage = neutron.db.migration.cli:main
        neutron-debug = neutron.debug.shell:main
        neutron-dhcp-agent = neutron.cmd.eventlet.agents.dhcp:main
        neutron-keepalived-state-change = neutron.cmd.keepalived_state_change:main
        neutron-ipset-cleanup = neutron.cmd.ipset_cleanup:main
        neutron-l3-agent = neutron.cmd.eventlet.agents.l3:main
        neutron-linuxbridge-agent = neutron.cmd.eventlet.plugins.linuxbridge_neutron_agent:main
        neutron-linuxbridge-cleanup = neutron.cmd.linuxbridge_cleanup:main
        neutron-macvtap-agent = neutron.cmd.eventlet.plugins.macvtap_neutron_agent:main
        neutron-metadata-agent = neutron.cmd.eventlet.agents.metadata:main
        neutron-netns-cleanup = neutron.cmd.netns_cleanup:main
        neutron-ns-metadata-proxy = neutron.cmd.eventlet.agents.metadata_proxy:main
        neutron-openvswitch-agent = neutron.cmd.eventlet.plugins.ovs_neutron_agent:main
        neutron-ovs-cleanup = neutron.cmd.ovs_cleanup:main
        neutron-pd-notify = neutron.cmd.pd_notify:main
        neutron-server = neutron.cmd.eventlet.server:main
        neutron-rpc-server = neutron.cmd.eventlet.server:main_rpc_eventlet
        ml2 = neutron.plugins.ml2.plugin:Ml2Plugin
    neutron.service_plugins =
        router = neutron.services.l3_router.l3_router_plugin:L3RouterPlugin
        firewall = neutron_fwaas.services.firewall.fwaas_plugin:FirewallPlugin
        lbaas = neutron_lbaas.services.loadbalancer.plugin:LoadBalancerPlugin
        vpnaas = neutron_vpnaas.services.vpn.plugin:VPNDriverPlugin
        metering = neutron.services.metering.metering_plugin:MeteringPlugin
        neutron.services.firewall.fwaas_plugin.FirewallPlugin = neutron_fwaas.services.firewall.fwaas_plugin:FirewallPlugin
        neutron.services.loadbalancer.plugin.LoadBalancerPlugin = neutron_lbaas.services.loadbalancer.plugin:LoadBalancerPlugin
        neutron.services.vpn.plugin.VPNDriverPlugin = neutron_vpnaas.services.vpn.plugin:VPNDriverPlugin
        qos = neutron.services.qos.qos_plugin:QoSPlugin
        bgp = neutron.services.bgp.bgp_plugin:BgpPlugin
        tag = neutron.services.tag.tag_plugin:TagPlugin
        flavors = neutron.services.flavors.flavors_plugin:FlavorsPlugin
        auto_allocate = neutron.services.auto_allocate.plugin:Plugin
        network_ip_availability = neutron.services.network_ip_availability.plugin:NetworkIPAvailabilityPlugin
        message_queue = neutron.services.qos.notification_drivers.message_queue:RpcQosServiceNotificationDriver
    neutron.ml2.type_drivers =
        flat = neutron.plugins.ml2.drivers.type_flat:FlatTypeDriver
        local = neutron.plugins.ml2.drivers.type_local:LocalTypeDriver
        vlan = neutron.plugins.ml2.drivers.type_vlan:VlanTypeDriver
        gre = neutron.plugins.ml2.drivers.type_gre:GreTypeDriver
        vxlan = neutron.plugins.ml2.drivers.type_vxlan:VxlanTypeDriver
    neutron.ml2.mechanism_drivers =

neutrton没有api进程，所有的api操作都是neutron-server处理，包括资源与插件的载入，数据库操作以及各个agent等。

def main():
        server.boot_server(_main_neutron_server)
    
    
    def _main_neutron_server():
        if cfg.CONF.web_framework == 'legacy':
            wsgi_eventlet.eventlet_wsgi_server()
        else:
            wsgi_pecan.pecan_wsgi_server()
    
    
    def main_rpc_eventlet():
        server.boot_server(rpc_eventlet.eventlet_rpc_server)

boot\_server主要是解析命令行指定的配置文件，--config-file=/etc/neutron/neutron.conf

启动服务，并调用\_main\_neutron\_server，\_main\_neutron\_server有两种 API方，启动wsgi\_server,传统的方式是通过eventlet，现在又新加pecan方式。默认情况下使用的eventlet方式，因此接着分析eventlet\_wsig\_server。

def eventlet_wsgi_server():
        neutron_api = service.serve_wsgi(service.NeutronApiService)
        start_api_and_rpc_workers(neutron_api)
    
    
    def start_api_and_rpc_workers(neutron_api):
        pool = eventlet.GreenPool()
    
        api_thread = pool.spawn(neutron_api.wait)
    
        try:
            neutron_rpc = service.serve_rpc()
        except NotImplementedError:
            LOG.info(_LI("RPC was already started in parent process by "
                         "plugin."))
        else:
            rpc_thread = pool.spawn(neutron_rpc.wait)
    
            plugin_workers = service.start_plugin_workers()
            for worker in plugin_workers:
                pool.spawn(worker.wait)
    
            # api and rpc should die together.  When one dies, kill the other.
            rpc_thread.link(lambda gt: api_thread.kill())
            api_thread.link(lambda gt: rpc_thread.kill())
    
        pool.waitall()

一部分是WSGI，另一部分就是rpc部分。这里将Netron提供的API功能封装成了NeutronApiService类

def serve_wsgi(cls):
    
        try:
            service = cls.create()
            service.start()
        except Exception:
            with excutils.save_and_reraise_exception():
                LOG.exception(_LE('Unrecoverable error: please check log '
                                  'for details.'))
    
        return service

NeutronApiService类方法"create"创建实例，start服务。

class NeutronApiService(WsgiService):
        """Class for neutron-api service.""" @classmethod
        def create(cls, app_name='neutron'):
    
            # Setup logging early, supplying both the CLI options and the
            # configuration mapping from the config file
            # We only update the conf dict for the verbose and debug
            # flags. Everything else must be set up in the conf file...
            # Log the options used when starting if we're in debug mode...
    
            config.setup_logging()
            service = cls(app_name)
            return service

NeutronApiService继承WsgiService，类方法create构造了实例

使用@staticmethod或@classmethod，就可以不需要实例化，直接类名.方法名()来调用，

*  @staticmethod不需要表示自身对象的self和自身类的cls参数，就跟使用函数一样。
 *  @classmethod也不需要self参数，但第一个参数需要是表示自身类的cls参数。

class WsgiService(object):
        """Base class for WSGI based services. For each api you define, you must also define these flags: :<api>_listen: The address on which to listen :<api>_listen_port: The port on which to listen """ def __init__(self, app_name):
            self.app_name = app_name
            self.wsgi_app = None
    
        def start(self):
            self.wsgi_app = _run_wsgi(self.app_name)
    
        def wait(self):
            self.wsgi_app.wait()

构造app\_name为neutron，start函数里真执行wsgi并运行服务：

def _run_wsgi(app_name):
        app = config.load_paste_app(app_name)
        if not app:
            LOG.error(_LE('No known API applications configured.'))
            return return run_wsgi_app(app)

load\_paste\_app加载paste定义的wsgi应用。

def load_paste_app(app_name):
        """Builds and returns a WSGI app from a paste config file. :param app_name: Name of the application to load """ loader = wsgi.Loader(cfg.CONF)
        app = loader.load_app(app_name)
        return app

wsgi.Loader从neutron.conf中读取deploy配置文件的路径，然后加载app，默认为/etc/neutron/api-paste.ini

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)

app的加载过程就是PasteDeploy的加载过程，PasteDeploy看作paste的一个扩展包，它主要是用来发现和配置WSGI应用。

配置文件api-paste.ini

[composite:neutron] use = egg:Paste#urlmap
    /: neutronversions
    /v2.0: neutronapi_v2_0
    
    [composite:neutronapi_v2_0] use = call:neutron.auth:pipeline_factory
    noauth = cors request_id catch_errors extensions neutronapiapp_v2_0
    keystone = cors request_id catch_errors authtoken keystonecontext extensions neutronapiapp_v2_0
    
    [filter:request_id] paste.filter_factory = oslo_middleware:RequestId.factory
    
    [filter:catch_errors] paste.filter_factory = oslo_middleware:CatchErrors.factory
    
    [filter:cors] paste.filter_factory = oslo_middleware.cors:filter_factory
    oslo_config_project = neutron
    
    [filter:keystonecontext] paste.filter_factory = neutron.auth:NeutronKeystoneContext.factory
    
    [filter:authtoken] paste.filter_factory = keystonemiddleware.auth_token:filter_factory
    
    [filter:extensions] paste.filter_factory = neutron.api.extensions:plugin_aware_extension_middleware_factory
    
    [app:neutronversions] paste.app_factory = neutron.api.versions:Versions.factory
    
    [app:neutronapiapp_v2_0] paste.app_factory = neutron.api.v2.router:APIRouter.factory

**由一个一个配置段section构成，每个section的格式如下:**

**\[type:name\]**

**其中，type包括以下几种:**

*  **应用:app,application.**
 *  **过滤器:filter,filte-app**
 *  **管道:pipeline**
 *  **组合:composite**

\[composite:neutron\] section,表示这是一个组合类型的配置，它的名字是neutron。组合类型表明它由若干WSGI应用构成。

use = egg:Paste#urlmap
    /: neutronversions
    /v2.0: neutronapi_v2_0

use 表明了使用Paste egg包中的paste.urlmap这个中间件的功能，功能是根据不通的URL前缀将请求路由给不同的WSGI应用。

对/的访问路由交给neutronversions处理

对/v2.0的网络路由交给neutronapi\_v2\_0处理

[app:neutronversions] paste.app_factory = neutron.api.versions:Versions.factory

工厂函数，指示了加载的模块和方法。

路径为：neutron/api/versions.py

class Versions(object):
    
        @classmethod
        def factory(cls, global_config, **local_config):
            return cls(app=None)
    
        @webob.dec.wsgify(RequestClass=wsgi.Request)
        def __call__(self, req):
            """Respond to a request for all Neutron API versions.""" version_objs = [
                {
                    "id": "v2.0",
                    "status": "CURRENT",
                },
            ]
    
            if req.path != '/':
                if self.app:
                    return req.get_response(self.app)
                language = req.best_match_language()
                msg = _('Unknown API version specified')
                msg = oslo_i18n.translate(msg, language)
                return webob.exc.HTTPNotFound(explanation=msg)
    
            builder = versions_view.get_view_builder(req)
            versions = [builder.build(version) for version in version_objs]
            response = dict(versions=versions)
            metadata = {}
    
            content_type = req.best_match_content_type()
            body = (wsgi.Serializer(metadata=metadata).
                    serialize(response, content_type))
    
            response = webob.Response()
            response.content_type = content_type
            response.body = wsgi.encode_body(body)
    
            return response
    
        def __init__(self, app):
            self.app = app

通过factory方法构造对象wsgi应用，处理对/的调用，@webob.dec.wsgify装饰器封装\_\_call\_\_

[composite:neutronapi_v2_0] use = call:neutron.auth:pipeline_factory
    noauth = cors request_id catch_errors extensions neutronapiapp_v2_0
    keystone = cors request_id catch_errors authtoken keystonecontext extensions neutronapiapp_v2_0

call后面是可调用对象

def pipeline_factory(loader, global_conf, **local_conf):
        """Create a paste pipeline based on the 'auth_strategy' config option.""" pipeline = local_conf[cfg.CONF.auth_strategy]
        pipeline = pipeline.split()
        filters = [loader.get_filter(n) for n in pipeline[:-1]]
        app = loader.get_app(pipeline[-1])
        filters.reverse()
        for filter in filters:
            app = filter(app)
        return app

从配置文件neutron.conf中读取auth\_strategy = keystone，从api-paste.ini中取到的pipeline为 **cors request\_id catch\_errors authtoken keystonecontext extensions neutronapiapp\_v2\_0**

**  先从pipeline中获取最后一个app为neutronapiapp\_v2\_0,从中加载app,然后依次用各个filter处理构造的app，并最终返回最后构造出的WSGI APP.**

**通过app\_factory工厂方法来构造app，然后通过不同的filter\_factory方法构造不同的filter对象，并将app依次通过filter对象处理。**

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

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()
            plugin = manager.NeutronManager.get_plugin()
            ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
            ext_mgr.extend_resources("2.0", attributes.RESOURCE_ATTRIBUTE_MAP)
    
            col_kwargs = dict(collection_actions=COLLECTION_ACTIONS,
                              member_actions=MEMBER_ACTIONS)

构造了一个APIRouter对象

routes\_mapper.Mapper:  用来构造了URL和对应controller映射，根据不同的URL路由给不同的controller处理。

manager.NeutronManager.get\_plugin: 根据配置加载核心插件（MLPlugin）

class NeutronManager(object):
        """Neutron's Manager class. Neutron's Manager class is responsible for parsing a config file and instantiating the correct plugin that concretely implements neutron_plugin_base class. The caller should make sure that NeutronManager is a singleton. """ _instance = None
    
        def __init__(self, options=None, config_file=None):
            # If no options have been provided, create an empty dict
            if not options:
                options = {}
    
            msg = validate_pre_plugin_load()
            if msg:
                LOG.critical(msg)
                raise Exception(msg)
    
            # NOTE(jkoelker) Testing for the subclass with the __subclasshook__
            #                breaks tach monitoring. It has been removed
            #                intentionally to allow v2 plugins to be monitored
            #                for performance metrics.
            plugin_provider = cfg.CONF.core_plugin
            LOG.info(_LI("Loading core plugin: %s"), plugin_provider)
            self.plugin = self._get_plugin_instance(CORE_PLUGINS_NAMESPACE,
                                                    plugin_provider)
            msg = validate_post_plugin_load()
            if msg:
                LOG.critical(msg)
                raise Exception(msg)
    
            # core plugin as a part of plugin collection simplifies
            # checking extensions
            # TODO(enikanorov): make core plugin the same as # the rest of service plugins
            self.service_plugins = {constants.CORE: self.plugin}
            self._load_service_plugins()

def _get_plugin_instance(self, namespace, plugin_provider):
        plugin_class = self.load_class_for_provider(namespace, plugin_provider)
        return plugin_class()

插件plugin\_provider: ml2 namespace: neutron.core\_plugins class: neutron.plugins.ml2.plugin.Ml2Plugin，并进行初始化，三个结构体：

Typemanager MechanismManager ExtensionManager

class TypeManager(stevedore.named.NamedExtensionManager):
        """Manage network segment types using drivers.""" def __init__(self):
            # Mapping from type name to DriverManager
            self.drivers = {}
    
            LOG.info(_LI("Configured type driver names: %s"),
                     cfg.CONF.ml2.type_drivers)
            super(TypeManager, self).__init__('neutron.ml2.type_drivers',
                                              cfg.CONF.ml2.type_drivers,
                                              invoke_on_load=True)
            LOG.info(_LI("Loaded type driver names: %s"), self.names())
            self._register_types()
            self._check_tenant_network_types(cfg.CONF.ml2.tenant_network_types)
            self._check_external_network_type(cfg.CONF.ml2.external_network_type)

根据配置文件/etc/neutron/plugin.ini中type\_drivers = flat,vxlan,vlan得到 cfg.CONF,ml2.type\_drivers为flat, vxlan, vlan

\_register\_types把三种写入self.drivers map中

def _load_service_plugins(self):
        """Loads service plugins. Starts from the core plugin and checks if it supports advanced services then loads classes provided in configuration. """ # load services from the core plugin first
        self._load_services_from_core_plugin()
    
        plugin_providers = cfg.CONF.service_plugins
        plugin_providers.extend(self._get_default_service_plugins())
        LOG.debug("Loading service plugins: %s", plugin_providers)

服务插件： 'router', 'metering', 'auto\_allocate', 'tag', 'timestamp\_core', 'network\_ip\_availability'

<table> 
 <tbody> 
  <tr> 
   <td>namespace</td> 
   <td>plugin</td> 
   <td>class</td> 
  </tr> 
  <tr> 
   <td>neutron.service_plugins</td> 
   <td>router</td> 
   <td>neutron.services.l3_router.l3_router_plugin.L3RouterPlugin</td> 
  </tr> 
  <tr> 
   <td>neutron.service_plugins</td> 
   <td>metering</td> 
   <td>neutron.services.metering.metering_plugin.MeteringPlugin</td> 
  </tr> 
  <tr> 
   <td>neutron.service_plugins</td> 
   <td>auto_allocate</td> 
   <td>neutron.services.auto_allocate.plugin.Plugin</td> 
  </tr> 
  <tr> 
   <td>neutron.service_plugins</td> 
   <td>tag</td> 
   <td>neutron.services.tag.tag_plugin.TagPlugin</td> 
  </tr> 
  <tr> 
   <td>neutron.service_plugins</td> 
   <td>network_ip_availability</td> 
   <td>neutron.services.network_ip_availability.plugin.NetworkIPAvailabilityPlugin</td> 
  </tr> 
 </tbody> 
</table>

col\_kwargs: \{'member\_actions': \['show', 'update', 'delete'\], 'collection\_actions': \['index', 'create'\]\}

for resource in RESOURCES:
        _map_resource(RESOURCES[resource], resource,
                      attributes.RESOURCE_ATTRIBUTE_MAP.get(
                          RESOURCES[resource], dict()))
        resource_registry.register_resource_by_name(resource)

RESOURCES = {
      'network': 'networks',
                 'subnet': 'subnets',
                 'subnetpool': 'subnetpools',
                 'port': 'ports'}

构造不同的URL的controller，networks subnets subnetpools ports

def _map_resource(collection, resource, params, parent=None):
        allow_bulk = cfg.CONF.allow_bulk
        allow_pagination = cfg.CONF.allow_pagination
        allow_sorting = cfg.CONF.allow_sorting
        controller = base.create_resource(
            collection, resource, plugin, params, allow_bulk=allow_bulk,
            parent=parent, allow_pagination=allow_pagination,
            allow_sorting=allow_sorting)
        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)

主要调用base.create\_resource

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实例化，

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)

关联处理函数，例如：get\_networkers get\_network create\_network update\_network delete\_network

wsgi\_resource.Resource所有的请求都会先交于resouce函数处理，进行反序列化和请求参数的获取，最终再交给controller处理

def resource(request):
        route_args = request.environ.get('wsgiorg.routing_args')
        if route_args:
            args = route_args[1].copy()
        else:
            args = {}
    
        # NOTE(jkoelker) by now the controller is already found, remove # it from the args if it is in the matchdict args.pop('controller', None)
        fmt = args.pop('format', None)
        action = args.pop('action', None)
        content_type = format_types.get(fmt, request.best_match_content_type())
        language = request.best_match_language()
        deserializer = deserializers.get(content_type)
        serializer = serializers.get(content_type)
    
        try:
            if request.body:
                args['body'] = deserializer.deserialize(request.body)['body']
    
            method = getattr(controller, action) result = method(request=request, **args)