Openstack ceilometer 宿主机监控模块扩展

首先，总体介绍Ceilometer采集监控数据到持久化存储的流程，在compute节点上通过pollster的轮询机制获取宿主机的各监控数据，通过publisher调用rpc将监控数据发送到消息队列，collector端根据约定的topic接收相应管道发送的数据，并调用storage接口 进行持久化存储，具体流程如下图所示。



其次，在该框架下如何扩展新的监控模块，顺其自然的想法即依次修改以上模块，其重点是配置文件的设置，需要定义发送端和接收端的topic、secrete和method才能将数据顺利发送到collector端进行存储。以下详细介绍各模块的修改方法。

一、compute数据采集pollster模块

该模块主要负责数据的采集，采集宿主机CPU、内存、网络等信息，封装成自定义数据格式后通过publisher发送到消息队列，该部分与ceilometer监控项扩展相似。 

1、在ceilometer/ceilometer/compute/plugin.py中定义宿主机获取数据的基类。 

class ServerPollster(plugin.PollsterBase):
@abc.abstractmethod
def get_samples(self, manager, cache, instances):

2、在ceilometer/ceilometer/compute下新建server_pollsters包，在该包下新建server_cpu.py、server_mem.py等。继承ServerPollster类实现get_samples方法，将获取的数据封装成定义的数据格式返回即可，可参考虚拟机采集数据架构。 

3、同监控项的扩展，在ceilometer/setup.cfg中增加监控项pollster的配置，最后所有修改完成后，重新安装ceilometer。 

二、发送数据publisher模块

1、修改ceilometer/pipeline.yaml配置文件，配置发送和接收数据的方式。

-
name: server_pipeline
interval: 60
counters:
- "server.cpu"
- "server.mem"
transformers:
publishers:
- rpc://?target=record_server_data&meter_type=server

其中，name为pipeline的名字；interval为轮询的时间间隔；counters为该pipeline的pollster名字；transformers若发送的数据在进行publisher之前需要进一步处理，可设置transformers进行转换；publishers指定发送数据到collector的方式，target指定collector端的接收方，根据meter_type获取rpc的topic、secret和method。 

同时，需要设置ceilometer/ceilometer.conf文件，设置server的topic、secret、method信息，具体配置如下： 

[publisher_rpc]
server_topic=server
server_secret=True
server_method=record_server_data

2、由于增加了新的topic发送通道，所以需要对ceilometer/ceilometer/publisher/rpc.py做相应的修改。首先，需要修改初始化函数，读取pipeline和ceilometer配置文件信息。其次，修改publish_counters函数，由于pipeline文件会对所有的counters每隔10分钟轮询一次，ceilometer默认采用topic为metering，其数据封装格式相同。但由于新增不同监控模块的数据封装格式不同，所以需要分开处理，10分钟轮询时只对counter类型为sample的数据进行publish。具体如下：

if self.meter_type == 'metering':
meters = [
meter_message_from_counter_beta(
counter,
self.topic_secret[self.meter_type],
source)
for counter in counters if type(counter) == sample.Sample
]
else:
meters = [
meter_message_from_counter_beta(
counter,
self.topic_secret[self.meter_type],
source)
for counter in counters
]
topic = self.meter_type

三、接收数据collector模块

1、ceilometer/ceilometer/collector/service.py文件initialize_service_hook中定义worker用于接收rpc发送过来的数据，具体配置如下：

for k, v in cfg.CONF.publisher_rpc.iteritems():
if k.endswith('topic'):
self.conn.create_worker(
v,
rpc_dispatcher.RpcDispatcher([self]),
'ceilometer.collector.' + v,
)

2、定义相应的target用于接收采集的数据，如下所示：

def record_server_data(self, context, data):
for dispatcher in self.dispatchers:
dispatcher.record_data(context, data, 'server')

其中，最后一个参数指定本次发送过来的topic类型，在调用dispatch.record_data方法时，用于调用相应的存储数据的方法，进而使得数据持久化到不同的数据库表中。 

3、dispatch也是在原基础上做了相应的调整，该调整根据当前数据接收的topic，调用不同的方法，进行数据库的存储，具体如下： 

def record_data(self, context, data, meter_type):
if not isinstance(data, list):
data = [data]
for meter in data:
if self.secret_method[meter_type] and
publisher_rpc.verify_signature(meter,
self.secret_method[meter_type][0]):
try:
if meter.get('timestamp'):
meter['timestamp'] =
self.time_to_date(meter['timestamp'])
method = getattr(self.storage_conn.__class__,
self.secret_method[meter_type][1])
method(self.storage_conn, meter)
except Exception as err:
LOG.error('Failed to record metering data: %s', err)
LOG.exception(err)
else:
LOG.warning(
'message signature invalid, discarding message: %r',
meter)

四、持久化存储模块

1、由于新增宿主机数据模块的监测，需增一张数据库表，所以storage模块也需做相应的修改。首先，需要修改ceilometer/ceilometer/storage/base.py文件，增加接口：

@abc.abstractmethod
def record_server_data(self, data):

2、在ceilometer/ceilometer/storage/impl_mongodb.py的Connection类中实现在接口。 

3、设计对应的数据库表结构，在ceilometer/ceilometer/storage/sqlalchemy/models.py中设置model对应的Server类。 

4、在ceilometer/ceilometer/storage/sqlalchemy/migrate_repo/versions/001_add_meter_table.py中定义对应的表。

以上配置完成后，source隔离环境重新安装ceilometer，重启服务，即可在mongodb数据库中查看到新增server表，表中存储对应counter的数据。