Ceilometer之polling agent代码分析

Ceilometer组件的功能是监控当前平台的运行情况，分为四大模块，polling agent,notification agent,collector和alarm，本文只介绍polling agent的代码分析，其他模块的代码分析将在以后给出。（devstack 版本是kilo）
程序入口在/ceilometer/cmd/polling.py，该目录下还有其他文件，都是相应功能的程序入口，打开该文件，我们可以看到如下内容
[mw_shl_code=applescript,true]def main_compute():
    service.prepare_service()
    os_service.launch(manager.AgentManager(['compute'])).wait()


# todo(dbelova): remove it someday. Needed for backward compatibility
def main_central():
    service.prepare_service()
    os_service.launch(manager.AgentManager(['central'])).wait()


def main_ipmi():
    service.prepare_service()
    os_service.launch(manager.AgentManager(['ipmi'])).wait()[/mw_shl_code]
我们可以看到启动了三个服务，对应的类是AgentManager，只是参数不同，我们以参数‘compute’的服务为例介绍。（compute服务主要获取计算节点的实例信息，central服务主要获取例如image，网络的信息，ipmi服务主要获取物理节点的资源信息）
该服务启动了manager.AgentManager，我们来到该类（ceilometer/agent/manager.py），看到如下内容
[mw_shl_code=applescript,true]class AgentManager(base.AgentManager):

    def __init__(self, namespaces=None, pollster_list=None):
        namespaces = namespaces or ['compute', 'central']
        pollster_list = pollster_list or []
        super(AgentManager, self).__init__(
            namespaces, pollster_list,
            group_prefix=cfg.CONF.polling.partitioning_group_prefix)[/mw_shl_code]
该类调用父类base.AgentManager的init方法，然后输入了三个参数namespaces,pollster_list和group_prefix，我们接着看父类的init方法，其中重要的如下
[mw_shl_code=applescript,true]extensions = (self._extensions('poll', namespace).extensions
                      for namespace in namespaces)
        if pollster_list:
            extensions = (itertools.ifilter(_match, exts)
                          for exts in extensions)

        self.extensions = list(itertools.chain(*list(extensions)))

        self.discovery_manager = self._extensions('discover')[/mw_shl_code]
这里用到了_extensions方法，返回一个ExtensionManager类（详细内容见http://docs.openstack.org/developer/stevedore/），该类调用一些plugin，plugin定义在setup.cfg（该文件很重要）中，比如以上代码中的extensions就是namespace为ceilometer.poll.compute的plugin，
[mw_shl_code=applescript,true]ceilometer.poll.compute =
    disk.read.requests = ceilometer.compute.pollsters.disk:ReadRequestsPollster
    disk.write.requests = ceilometer.compute.pollsters.disk:WriteRequestsPollster
    disk.read.bytes = ceilometer.compute.pollsters.disk:ReadBytesPollster
    disk.write.bytes = ceilometer.compute.pollsters.disk:WriteBytesPollster
    disk.read.requests.rate = ceilometer.compute.pollsters.disk:ReadRequestsRatePollster
    disk.write.requests.rate = ceilometer.compute.pollsters.disk:WriteRequestsRatePollster
    disk.read.bytes.rate = ceilometer.compute.pollsters.disk:ReadBytesRatePollster
#还有很多[/mw_shl_code]

以上plugin就是获取相应的实例信息，比如cpu，内存等，相应的discovery_manager也类似，只不过它是为extensions提供一些必要的资源，比如为统计cpu的pollster提供节点的所有实例。
接着就是最重要的方法了——start，核心代码如下
[mw_shl_code=applescript,true]for interval, task in six.iteritems(self.setup_polling_tasks()):
            delay_time = (interval + delay_polling_time if delay_start
                          else delay_polling_time)
            self.tg.add_timer(interval,
                              self.interval_task,
                              initial_delay=delay_time,
                              task=task)[/mw_shl_code]
首先是第一行的self.setup_polling_tasks()，该方法读取pipeline，并与extensions比较，将pipeline和相应的plugins对应起来，那么pipeline是什么呢，是/etc/ceilometer/pipeline.yaml
[mw_shl_code=applescript,true] - name: cpu_source
      interval: 20
      meters:
          - "cpu"
      sinks:
          - cpu_sink

- name: cpu_sink
      transformers:
          - name: "rate_of_change"
            parameters:
                target:
                    name: "cpu_util"
                    unit: "%"
                    type: "gauge"
                    scale: "100.0 / (10**9 * (resource_metadata.cpu_number or 1))"
      publishers:
          - rpc://?name=hqc&age=20
[/mw_shl_code]






补充内容 (2016-3-3 09:35):
以上代码就是pipeline的一个例子，具体解释见楼下

pipeline分为两部分，一部分是pipeline的相关信息，另一部分是sink，pipeline name指定要获取那些资源（meters），时间间隔（interval）等，sink指定获取的信息如何处理，一般分为两部分，transform和publish。关于transform和publish我稍后继续，现在我们回到start方法，当获取了pipeline和plugin的对应关系后，遍历polling_tasks，获得task和interval。请看如下代码[mw_shl_code=applescript,true]self.tg.add_timer(interval,
                              self.interval_task,
                              initial_delay=delay_time,
                              task=task)[/mw_shl_code]

意思是定时执行self.interval_task方法，该方法就是poll相关信息并publish。poll就是执行相关plugin以获取信息，publish有多种方式，官方定i如下
[mw_shl_code=applescript,true]notifier
It can be specified in the form of notifier://?option1=value1&option2=value2. It emits data over AMQP using oslo.messaging. This is the recommended method of publishing.
rpc
It can be specified in the form of rpc://?option1=value1&option2=value2. It emits metering data over lossy AMQP. This method is synchronous and may experience performance issues. This publisher is deprecated in Liberty in favor of the notifier publisher.
udp
It can be specified in the form of udp://<host>:<port>/. It emits metering data for over UDP.
file
It can be specified in the form of file://path?option1=value1&option2=value2. This publisher records metering data into a file.[/mw_shl_code]
分为notifier,rpc,udp和file，file比较简单不赘述，这里详细介绍下rpc和notifier。

请看楼下

心心苦苦写的东西，提交失败，然后就没有了，f  u  c  k,没心情写了
notifier：默认流程如下
ceilometer.publisher.messaging:SampleNotifierPublisher
NotifierPublisher
oslo.messaging.Notifier.sample方法
oslo.messaging.Notifier._notify方法，该方法调用namespace为oslo.messaging.notify.drivers，name是messagingv2的plugin。

最后信息是发送到exchange为ceilometer，queue是meter.sample,topic是meter.sample，可以自己写个小程序获取


rpc：将消息进行rpc调用，若没有指定target，默认执行rpc server的record_metering_data方法，最后存入ceilometer数据库的sample表中，自定义rpc的格式如下rpc://?target=fuck_about&tool=dick

参考文献：1.本版本贴之前的文章
2.http://docs.openstack.org/admin- ... try-data-collection
3.http://docs.openstack.org/admin- ... val.html#publishers
4.http://docs.openstack.org/developer/ceilometer/

关于pipeline中的transform，也是查看setup.cfg文件，找到相关类[mw_shl_code=applescript,true]ceilometer.transformer =
    accumulator = ceilometer.transformer.accumulator:TransformerAccumulator
    unit_conversion = ceilometer.transformer.conversions:ScalingTransformer
    rate_of_change = ceilometer.transformer.conversions:RateOfChangeTransformer
    aggregator = ceilometer.transformer.conversions:AggregatorTransformer
    arithmetic = ceilometer.transformer.arithmetic:ArithmeticTransformer[/mw_shl_code]

transformer有不同的name，不同的name对应不同的方法