OpenStack 云平台流量监控插件tap-as-a-service(Taas)代码解析(二)：

在上一篇文章中，以create_tap_service为例，讲解了OpenStack中云端流量捕获插件Tap-as-a-service的Plugin的代码流程

（https://blog.csdn.net/m0_37313888/article/details/82693789）

先回顾一下Taas的结构，这篇文章我将继续讲解Taas Agent与Taas Driver的工作流程

（neutron对应的OpenStack版本：Queuens,关于如何在生产环境中的openstack上安装tap-as-a-service插件请看我的另一篇文章https://blog.csdn.net/m0_37313888/article/details/83450245）

1.Taas插件加载与初始化

众所周知，OpenStack中Plugin只负责数据库以及消息队列的维护，工作具体是在agent节点上执行的。taas作为在这里是作为OpenvSwitch的一个extension集成到neutron-openvswitch-agent中,并且在neutron-openvswitch-agent启动时进行加载与初始化。

首先看一看taas 在agent上的初始化过程

在neutron-openvswitch-agent加载时，顺便把taas插件加载的过程：

1）插件加载与实例化

在neutron-openvswitch-agent初始化代码中(代码位置：neutron/plugins/ml2/driver/openvswitch/agent/ovs_neutron_agent.py)

,第2290行附近，有这么一段代码，这段代码将实例从配置文件中读出并且添加到ext_mgr作为ext_mgr的一个成员对象。

2288    ext_manager.register_opts(cfg.CONF)
2289
2290    ext_mgr = ext_manager.L2AgentExtensionsManager(cfg.CONF)

对应到neutron/agent/l2/l2_agent_extensions_manager.py中的下面这段代码

21    L2_AGENT_EXT_MANAGER_NAMESPACE = 'neutron.agent.l2.extensions'
22
23
24    def register_opts(conf):
25        agent_ext_mgr_config.register_agent_ext_manager_opts(conf)

neutron/conf/agent/agent_extensions_manager.py

17    AGENT_EXT_MANAGER_OPTS = [
18        cfg.ListOpt('extensions',
19                    default=[],
20                    help=_('Extensions list to use')),
21    ]
22
23
24    def register_agent_ext_manager_opts(cfg=cfg.CONF):
25        cfg.register_opts(AGENT_EXT_MANAGER_OPTS, 'agent')

代码是直接读了配置文件中agent部分的extensions选项。具体的配置过程请阅读https://blog.csdn.net/m0_37313888/article/details/83450245

加载插件的具体位置如下,加载下面这个插件就是创建了下面这个类的一个实例。

[neutron.agent.l2.extensions]
taas = neutron_taas.services.taas.agents.extensions.taas:TaasAgentExtension

2）插件的初始化

插件的加载只是在内存中生成了一个实例。但是这个插件实例还没有具体发挥它的作用,这个插件实例在OVSNeutronAgent的运行过程中才真正发挥它的作用。在neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py中插件管理器作为了OVSNeutronAgent的成员参与到了OVSNeutronAgent的工作中.(下图中的ext_mgr就是包含了taas的插件管理器)

neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py

2297    try:
2298        agent = OVSNeutronAgent(bridge_classes, ext_mgr, cfg.CONF)
2299        capabilities.notify_init_event(n_const.AGENT_TYPE_OVS, agent)
2300    except (RuntimeError, ValueError) as e:
2301        LOG.error("%s Agent terminated!", e)
2302        sys.exit(1)
2303    agent.daemon_loop()

插件管理器被传递到OVSNeutronAgent的成员变量中（下图）

140    self.ext_manager = ext_manager

之后，还是在这个代码中，进行下面的操作：1.绑定agent_api操作2.初始化所有的插件

219        agent_api = ovs_ext_api.OVSAgentExtensionAPI(self.int_br, self.tun_br)
220        self.ext_manager.initialize(
221            self.connection, constants.EXTENSION_DRIVER_TYPE, agent_api)

查看具体的代码实现如下：

class AgentExtensionsManager(stevedore.named.NamedExtensionManager):"""Manage agent extensions."""def __init__(self, conf, namespace):super(AgentExtensionsManager, self).__init__(namespace, conf.agent.extensions,invoke_on_load=True, name_order=True)LOG.info("Loaded agent extensions: %s", self.names())def initialize(self, connection, driver_type, agent_api=None):# Initialize each agent extension in the list.for extension in self:LOG.info("Initializing agent extension '%s'", extension.name)# If the agent has provided an agent_api object, this object will# be passed to all interested extensions.  This object must be# consumed by each such extension before the extension's# initialize() method is called, as the initialization step# relies on the agent_api already being available.extension.obj.consume_api(agent_api)extension.obj.initialize(connection, driver_type)

taas-agent-extension在这段代码中被加载，我们来看看taas-agent-extension做了啥

class TaasAgentExtension(l2_extension.L2AgentExtension):def initialize(self, connection, driver_type):"""Initialize agent extension."""self.taas_agent = taas_ovs_agent.TaasOvsAgentRpcCallback(cfg.CONF, driver_type)self.taas_agent.consume_api(self.agent_api)self.taas_agent.initialize()def consume_api(self, agent_api):"""Receive neutron agent API objectAllows an extension to gain access to resources internal to theneutron agent and otherwise unavailable to the extension."""self.agent_api = agent_apidef handle_port(self, context, port):passdef delete_port(self, context, port):pass

三件事：

1. 加载本地rpc回调函数类: TaasOvsAgentRpcCallback

2. 加载agent_api: ovs_ext_api.OVSAgentExtensionAPI(self.int_br, self.tun_br)

3. 初始化本地rpc回调函数类

本地的rpc回调函数类如下：

class TaasOvsAgentRpcCallback(api.TaasAgentRpcCallbackMixin):def __init__(self, conf, driver_type):LOG.debug("TaaS OVS Agent initialize called")self.conf = confself.driver_type = driver_typesuper(TaasOvsAgentRpcCallback, self).__init__()def initialize(self):self.taas_driver = manager.NeutronManager.load_class_for_provider('neutron_taas.taas.agent_drivers', self.driver_type)()self.taas_driver.consume_api(self.agent_api)self.taas_driver.initialize()self._taas_rpc_setup()TaasOvsAgentService(self).start()def consume_api(self, agent_api):self.agent_api = agent_apidef _invoke_driver_for_plugin_api(self, context, args, func_name):........try:self.taas_driver.__getattribute__(func_name)(args)except Exception:LOG.debug("Failed to invoke the driver")returndef create_tap_service(self, context, tap_service, host):........def create_tap_flow(self, context, tap_flow_msg, host):........def delete_tap_service(self, context, tap_service, host):........def delete_tap_flow(self, context, tap_flow_msg, host):........def _taas_rpc_setup(self):# setup RPC to msg taas pluginself.taas_plugin_rpc = TaasOvsPluginApi(topics.TAAS_PLUGIN, self.conf.host)endpoints = [self]conn = n_rpc.Connection()conn.create_consumer(topics.TAAS_AGENT, endpoints, fanout=False)conn.consume_in_threads()def periodic_tasks(self):........

我们来重点看一看上面函数中的initialize()部分的代码，这个是agent上的taas-extension的rpc代码的初始化的核心代码

self.taas_driver = manager.NeutronManager.load_class_for_provider('neutron_taas.taas.agent_drivers', self.driver_type)()

看一看NeutronManager.load_class_for_provider

    @staticmethoddef load_class_for_provider(namespace, plugin_provider):"""Loads plugin using alias or class name:param namespace: namespace where alias is defined:param plugin_provider: plugin alias or class name:returns: plugin that is loaded:raises ImportError: if fails to load plugin"""try:return runtime.load_class_by_alias_or_classname(namespace,plugin_provider)except ImportError:with excutils.save_and_reraise_exception():LOG.error("Plugin '%s' not found.", plugin_provider)

runtime.load_class_by_alias_or_classname(namespace, plugin_provider)如下:

def load_class_by_alias_or_classname(namespace, name):if not name:LOG.error("Alias or class name is not set")raise ImportError(_("Class not found."))try:mgr = driver.DriverManager(namespace, name, warn_on_missing_entrypoint=False)class_to_load = mgr.driverexcept RuntimeError:e1_info = sys.exc_info()# Fallback to class nametry:class_to_load = importutils.import_class(name)except (ImportError, ValueError):LOG.error("Error loading class by alias",exc_info=e1_info)LOG.error("Error loading class by class name",exc_info=True)raise ImportError(_("Class not found."))return class_to_load

通过stevedore.driver.DriverManager类动态加载模块。这是一个与动态加载类有关的模块。关于这个模块的使用，可以看看这篇文章：http://www.360doc.com/content/14/0429/19/9482_373285413.shtml

然后self.taas_driver通过给出的namespace:

neutron_taas.taas.agent_drivers

与 driver_type:

ovs 被加载。具体被加载的类写在了setup.cfg的entry_points里面

[entry_points]
neutron.agent.l2.extensions =taas = neutron_taas.services.taas.agents.extensions.taas:TaasAgentExtension
neutron_taas.taas.agent_drivers =ovs = neutron_taas.services.taas.drivers.linux.ovs_taas:OvsTaasDriver
neutron.service_plugins =taas = neutron_taas.services.taas.taas_plugin:TaasPlugin
neutron.db.alembic_migrations =tap-as-a-service = neutron_taas.db.migration:alembic_migration
tempest.test_plugins =tap-as-a-service = neutron_taas.tests.tempest_plugin.plugin:NeutronTaaSPlugin
neutronclient.extension =tap_service = neutron_taas.taas_client.tapservicetap_flow = neutron_taas.taas_client.tapflow

即，self.taas_driver =

neutron_taas.services.taas.drivers.linux.ovs_taas:OvsTaasDriver

self.taas_driver.consume_api(self.agent_api)

在self.taas_driver, 即neutron_taas.services.taas.drivers.linux.ovs_taas:OvsTaasDriver中，consume_api()方法如下：

    def consume_api(self, agent_api):self.agent_api = agent_api

简单的增加了一个self.agent_api成员

self.taas_driver.initialize()

还是看一下self.taas_driver,

    def initialize(self):self.int_br = self.agent_api.request_int_br()self.tun_br = self.agent_api.request_tun_br()self.tap_br = OVSBridge_tap_extension('br-tap', self.root_helper)# Prepare OVS bridges for TaaSself.setup_ovs_bridges()# Setup key-value manager for ingress BCMC flowsself.bcmc_kvm = taas_ovs_utils.key_value_mgr(4096)

这个时候agent_api就派上用场了！这个对象，顾名思义，就是一个agent所映射产生的一个api对象，提供许多访问本地资源的接口，在ovs中，是本地所建立的网桥与相关网络设备。我们看到，先是获取了int-br网桥与tun_br 网桥，来看一看获取网桥的基本操作，这里agent_api是ovs_ext_api.OVSAgentExtensionAPI(self.int_br, self.tun_br)

如下

class OVSAgentExtensionAPI(object):'''Implements the Agent API for Open vSwitch agent.Extensions can gain access to this API by overriding the consume_apimethod which has been added to the AgentExtension class.'''def __init__(self, int_br, tun_br):super(OVSAgentExtensionAPI, self).__init__()self.br_int = int_brself.br_tun = tun_brdef request_int_br(self):"""Allows extensions to request an integration bridge to use forextension specific flows."""return OVSCookieBridge(self.br_int)def request_tun_br(self):"""Allows extensions to request a tunnel bridge to use forextension specific flows.If tunneling is not enabled, this method will return None."""if not self.br_tun:return Nonereturn OVSCookieBridge(self.br_tun)

这里OVSCookieBridge让我们想到了web开发中的cookie,但是这个其实就是把本地的网桥信息浅拷贝一份然后返回给程序进一步使用

class OVSCookieBridge(object):def __new__(cls, bridge):cookie_bridge = bridge.clone()cookie_bridge.set_agent_uuid_stamp(bridge.request_cookie())return cookie_bridgedef __init__(self, bridge):pass

    def clone(self):'''Used by OVSCookieBridge, can be overridden by subclasses if abehavior different from copy.copy is needed.'''return copy.copy(self)

接着，然后建立了一个tap网桥。这一步是在agent上实现流量拷贝、迁移、以及重定向的重头戏。

self.setup_ovs_bridges()

具体代码在neutron_taas/services/taas/drivers/linux/ovs_taas.py

这里为了直观的展示这个流程，把流程分为4个部分

1.创建一个tap网桥

刚开始我看这一段的时候也很是迷惑，因为ovs有专门的流量镜像功能啊，为何不直接管理ovs做一个流量镜像呢？原来，tap-as--a-service的主要设计理念是为不同节点上的虚拟机提供统一的流量捕获服务，这样的话方案中流量必须能够方便的进行跨主机迁移。如果采用vxlan隧道的话就很容易建立新的流量隧道实现流量的跨节点迁移。换句话说，流量既能够定向到相同宿主机上的另一台虚拟机，也能重定向到另一台宿主机上的另一台虚拟机！

        self.tap_br.create()# Connect br-tap to br-int and br-tunself.int_br.add_patch_port('patch-int-tap', 'patch-tap-int')self.tap_br.add_patch_port('patch-tap-int', 'patch-int-tap')self.tun_br.add_patch_port('patch-tun-tap', 'patch-tap-tun')self.tap_br.add_patch_port('patch-tap-tun', 'patch-tun-tap')# Get patch port IDspatch_tap_int_id = self.tap_br.get_port_ofport('patch-tap-int')patch_tap_tun_id = self.tap_br.get_port_ofport('patch-tap-tun')patch_tun_tap_id = self.tun_br.get_port_ofport('patch-tun-tap')

2.删除预先定义的与tap-as-a-service相关的所有规则

        self.tap_br.delete_flows(table=0)self.tap_br.delete_flows(table=taas_ovs_consts.TAAS_RECV_LOC)self.tap_br.delete_flows(table=taas_ovs_consts.TAAS_RECV_REM)self.tun_br.delete_flows(table=0,in_port=patch_tun_tap_id)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_SEND_UCAST)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_SEND_FLOOD)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_CLASSIFY)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_DST_CHECK)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_SRC_CHECK)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_DST_RESPOND)self.tun_br.delete_flows(table=taas_ovs_consts.TAAS_SRC_RESPOND)

3.初始化tap网桥的OpenFlow流表

        self.tap_br.add_flow(table=0,priority=1,in_port=patch_tap_int_id,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_RECV_LOC)self.tap_br.add_flow(table=0,priority=1,in_port=patch_tap_tun_id,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_RECV_REM)self.tap_br.add_flow(table=0,priority=0,actions="drop")self.tap_br.add_flow(table=taas_ovs_consts.TAAS_RECV_LOC,priority=0,actions="output:%s" % str(patch_tap_tun_id))self.tap_br.add_flow(table=taas_ovs_consts.TAAS_RECV_REM,priority=0,actions="drop")

这段代码可能看起来比较吃力，我在下面做出一些解释:

里面出现了 taas_ovs_consts.TAAS_RECV_LOC 与 taas_ovs_consts.TAAS_RECV_REM两张表。结合后面的代码仔细一看，TAAS_RECV_LOC负责处理从br-int过来的流量（流出vm的流量），默认为转发到br-tun;TAAS_RECV_REM负责处理从br-tun过来的流量，默认为不让任何来自于br-tun的流量通过。来自其他网桥的流量也默认不通过。

4.初始化br-tun网桥的OpenFlow流表

        self.tun_br.add_flow(table=0,priority=1,in_port=patch_tun_tap_id,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_SEND_UCAST)self.tun_br.add_flow(table=taas_ovs_consts.TAAS_SEND_UCAST,priority=0,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_SEND_FLOOD)flow_action = self._create_tunnel_flood_flow_action()if flow_action != "":self.tun_br.add_flow(table=taas_ovs_consts.TAAS_SEND_FLOOD,priority=0,actions=flow_action)self.tun_br.add_flow(table=taas_ovs_consts.TAAS_CLASSIFY,priority=2,reg0=0,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_DST_CHECK)self.tun_br.add_flow(table=taas_ovs_consts.TAAS_CLASSIFY,priority=1,reg0=1,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_DST_CHECK)self.tun_br.add_flow(table=taas_ovs_consts.TAAS_CLASSIFY,priority=1,reg0=2,actions="resubmit(,%s)" %taas_ovs_consts.TAAS_SRC_CHECK)self.tun_br.add_flow(table=taas_ovs_consts.TAAS_DST_CHECK,priority=0,actions="drop")self.tun_br.add_flow(table=taas_ovs_consts.TAAS_SRC_CHECK,priority=0,actions="drop")self.tun_br.add_flow(table=taas_ovs_consts.TAAS_DST_RESPOND,priority=2,reg0=0,actions="output:%s" % str(patch_tun_tap_id))self.tun_br.add_flow(table=taas_ovs_consts.TAAS_DST_RESPOND,priority=1,reg0=1,actions=("output:%s,""move:NXM_OF_VLAN_TCI[0..11]->NXM_NX_TUN_ID""[0..11],mod_vlan_vid:2,output:in_port" %str(patch_tun_tap_id)))self.tun_br.add_flow(table=taas_ovs_consts.TAAS_SRC_RESPOND,priority=1,actions=("learn(table=%s,hard_timeout=60,""priority=1,NXM_OF_VLAN_TCI[0..11],""load:NXM_OF_VLAN_TCI[0..11]->NXM_NX_TUN_ID""[0..11],load:0->NXM_OF_VLAN_TCI[0..11],""output:NXM_OF_IN_PORT[])" %taas_ovs_consts.TAAS_SEND_UCAST))return

以上就是给计算节点开机后，taas插件的初始化过程。关于在实际使用时流表的配置，以及为什么要这么配置，我后期会专门写一篇文章来讲解。

OpenStack 云平台流量监控插件tap-as-a-service(Taas)代码解析(二)：相关推荐

OpenStack 云平台流量监控插件tap-as-a-service(Taas)源码解析(一)：
由于最近在整合一个云端流量捕获插件,所以特意的把OpenStack中的Tap-as-a-service的源代码拿来学习.项目的具体文件在github上.https://github.com/opens ...
OpenStack 云平台流量监控插件tap-as-a-service(Taas)安装步骤（OpenStack queens版本，非devstack）
网上貌似没有任何相关的博客....下面我来讲一讲怎么将devstack下的openstack插件Tap-as-a-service 集成到手工搭建的openstack上首先,github上面放的都是d ...
Openstack云平台脚本部署之Ceph存储集群配置（十四）
目录一.简介二.部署脚本三.参考文档四.源码五.系列文章一.简介 Openstack的后端存储对接Ceph分布式统一存储,将Glance镜像.Nova虚拟机及快照.Cinder-Volum ...
使用(telemetry)ceilometer+gnocchi对openstack云平台监控数据采集和处理的实践与优化
使用(telemetry)ceilometer+gnocchi对openstack云平台监控数据采集和处理的实践与优化. 在做openstack资源统计监控中通过 (telemetry)ceilome ...
openStack 云平台管理节点管理网口流量非常大出现丢包严重终端总是时常中断问题调试及当前测试较有效方案...
openStack 云平台管理节点管理网口流量非常大出现丢包严重终端总是时常中断问题调试及当前测试较有效方案 tuning for Data Transfer hosts connected at ...
Openstack云平台的搭建与部署（具体实验过程截图评论拿）
目录 Openstack云平台的搭建与部署................................... 3 Keywords:Openstack.Cloud Computing.Iaas.. ...
CentOS上搭建OpenStack云平台只需8G内存!!!
Hello!大家好,今天阿坤为大家带来8G内存的云平台搭建,按照官方的标准是需要32G,最少也是16G内存的,但是阿坤觉的还不是极限,所以今天带大家挑战8G内存搭建最小云平台! 环境准备 1.总共需要 ...
手把手教你搭建OpenStack云平台（超级详细）
一.前言 OpenStack云平台搭建需要两个节点,一个是controller(控制节点),另一个是compute(计算节点). 控制节点(controller)规划如下: 一块200G的硬盘.两块网 ...
云计算 openstack 云平台搭建详细教程（基于 Vmware 虚拟机搭建）
OpenStack 和云计算 1.OpenStack 基本概述早在1988年,类似云计算概念的"网络就是计算机"概念就被 SUN 微系统公司的合作创建者约翰 · 盖奇首次提出 ...

OpenStack 云平台流量监控插件tap-as-a-service(Taas)代码解析(二)：

OpenStack 云平台流量监控插件tap-as-a-service(Taas)代码解析(二)：相关推荐

最新文章

热门文章