Ceilometer项目源码分析----ceilometer-agent-compute服务的初始化和启动

感谢朋友支持本博客，欢迎共同探讨交流，由于能力和时间有限，错误之处在所难免，欢迎指正！如果转载，请保留作者信息。
博客地址：http://blog.csdn.net/gaoxingnengjisuan
邮箱地址：dong.liu@siat.ac.cnPS：最近没有登录博客，很多朋友的留言没有看见，这里道歉！还有就是本人较少上QQ，可以邮件交流。
ceilometer-agent-compute服务的初始化和启动
本篇博客将解析服务组件ceilometer-agent-compute的初始化和启动操作,ceilometer-agent-compute服务组件主要用来收集计算节点上的虚拟机实例的监控数据，在每一个计算节点上都要运行这个服务组件，该agent通过Stevedore管理了一组pollster插件，分别用来获取计算节点上虚拟机的CPU,Disk IO,Network IO,Instance这些信息，这些信息大部分是通过调用Hypervisor的API来获取的，需要定期Poll轮询收集信息。
来看方法/ceilometer/cli.py----def agent_compute,这个方法即实现了ceilometer-agent-compute服务的初始化和启动操作。

def agent_compute():
    """
    加载并启动指定的服务compute_manager.AgentManager()；
    ceilometer.poll.compute：
    该组件用来收集计算节点上的信息，在每一个计算节点上都要运行一个Compute Agent，
    该Agent通过Stevedore管理了一组pollster插件，分别用来获取虚拟机的CPU,Disk IO,
    Network IO,Instance这些信息，值得一提的是这些信息大部分是通过调用Hypervisor的API
    来获取的，目前，Ceilometer提供了Libvirt的API和Hyper-V的API。
    """
    service.prepare_service()
    os_service.launch(compute_manager.AgentManager()).wait()

[b]1 服务ceilometer-agent-compute的初始化操作[/b]服务ceilometer-agent-compute的初始化操作主要实现了以下内容的操作：（1）根据指定参数获取命名空间ceilometer.poll.compute，获取与ceilometer.poll.compute相匹配的所有插件，并加载；ceilometer.poll.compute所指定的插件描述了如何获取收集所监控的虚拟机实例相关的监控采样数据；
（2）根据指定参数获取命名空间ceilometer.discover，获取与ceilometer.discover相匹配的所有插件，并加载；ceilometer.discover所指定的插件描述了如何发现主机上的监控的虚拟机。
ceilometer.discover = local_instances = ceilometer.compute.discovery:InstanceDiscovery
（3）获取管理员操作的上下文环境类的初始化对象；
（4）建立线程池，用于后续服务中若干操作的运行；（5）加载命名空间ceilometer.compute.virt，描述了获取虚拟机实例的信息（实例数据，CPU数据，网卡数据和磁盘数据等）的方式；class AgentManager(agent.AgentManager)----def __init__

class AgentManager(agent.AgentManager):
    def __init__(self):
       
        super(AgentManager, self).__init__('compute', ['local_instances'])      
        """
        定义获取虚拟机实例的信息（实例数据，CPU数据，网卡数据和磁盘数据等）的方式；
        加载命名空间ceilometer.compute.virt；
        ceilometer.compute.virt =
            libvirt = ceilometer.compute.virt.libvirt.inspector:LibvirtInspector
            hyperv = ceilometer.compute.virt.hyperv.inspector:HyperVInspector
            vsphere = ceilometer.compute.virt.vmware.inspector:VsphereInspector
        """
        self._inspector = virt_inspector.get_hypervisor_inspector()

class AgentManager(os_service.Service)----def __init__

class AgentManager(os_service.Service):

    def __init__(self, namespace, default_discovery=[]):
        super(AgentManager, self).__init__()
        self.default_discovery = default_discovery
        """
        加载命名空间ceilometer.poll.compute的所有插件：
        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
        disk.write.bytes.rate = ceilometer.compute.pollsters.disk:WriteBytesRatePollster
        cpu = ceilometer.compute.pollsters.cpu:CPUPollster
        cpu_util = ceilometer.compute.pollsters.cpu:CPUUtilPollster
        network.incoming.bytes = ceilometer.compute.pollsters.net:IncomingBytesPollster
        network.incoming.packets = ceilometer.compute.pollsters.net:IncomingPacketsPollster
        network.outgoing.bytes = ceilometer.compute.pollsters.net:OutgoingBytesPollster
        network.outgoing.packets = ceilometer.compute.pollsters.net:OutgoingPacketsPollster
        network.incoming.bytes.rate = ceilometer.compute.pollsters.net:IncomingBytesRatePollster
        network.outgoing.bytes.rate = ceilometer.compute.pollsters.net:OutgoingBytesRatePollster
        instance = ceilometer.compute.pollsters.instance:InstancePollster
        instance_flavor = ceilometer.compute.pollsters.instance:InstanceFlavorPollster
        memory.usage = ceilometer.compute.pollsters.memory:MemoryUsagePollster
        """
        self.pollster_manager = self._extensions('poll', namespace)

        """
        加载命名空间所有插件ceilometer.discover：
        ceilometer.discover =
        local_instances = ceilometer.compute.discovery:InstanceDiscovery
        """
        self.discovery_manager = self._extensions('discover')
        self.context = context.RequestContext('admin', 'admin', is_admin=True)

class AgentManager(os_service.Service)----def _extensions

def _extensions(category, agent_ns=None):  
        """ 
        根据指定参数获取命名空间namespace， 
        获取与namespace相匹配的所有插件，并加载； 
        """  
        namespace = ('ceilometer.%s.%s' % (category, agent_ns) if agent_ns else 'ceilometer.%s' % category)  
              
        """ 
        获取与namespace相匹配的所有插件，并加载； 
        """  
        return extension.ExtensionManager(  
            namespace=namespace,  
            invoke_on_load=True,  
        )

class Service(object)----def __init__

class Service(object):  
        """Service object for binaries running on hosts."""  
        def __init__(self, threads=1000):  
            self.tg = threadgroup.ThreadGroup(threads)  
      
            # signal that the service is done shutting itself down:  
            self._done = event.Event()

[b][b]2 服务[b][b]ceilometer-agent-compute[/b]的启动操作[/b][/b][/b][b][/b]这里服务ceilometer-agent-compute与服务ceilometer-agent-central的启动操作的实现代码是完全一致的，只不过所获取的监控任务不同而已，所调用的采集监控项采样数据的方法不同而已；服务ceilometer-agent-compute的启动操作周期性地实现以下任务：
（1）遍历任务（通道），获取每个任务指定获取的监控项的采样数据；
（2）针对每个监控项的采样数据，实现发布监控项采样数据样本到消息队列，其中实现采样数据发布的方式有三种，即RPC/UDP/FILE； 其中，RPC将会发布相关消息到消息队列，后续的collector组件服务将会监听相应的消息队列来获取这些数据信息；UDP将会建立socket建立一个信息通道，实现发送相关消息数据，而后续的collector组件服务将会通过这个信息通道接收相关的消息数据；FILE将会直接保存相关消息数据到指定的日志文件中。[b]class AgentManager(os_service.Service)----def start[/b]

class AgentManager(os_service.Service):  
    def start(self):  
        self.pipeline_manager = pipeline.setup_pipeline()  
        for interval,task in self.setup_polling_tasks().iteritems():  
            self.tg.add_timer(interval,  
                              self.interval_task,  
                              task=task)

class AgentManager(os_service.Service)----def interval_task

def interval_task(task):  
    task.poll_and_publish()

class PollingTask(object)----def poll_and_publish

class AgentManager(os_service.Service):
    def poll_and_publish(self):
        """
        创建轮徇的任务；
        任务以一定时间间隔周期性地进行；
        
        遍历任务（通道），获取每个任务指定获取的监控项的采样数据；
        针对每个监控项的采样数据，实现发布监控项采样数据样本到消息队列；
        """
        """
        通过nova客户端访问nova数据库，获取本地主机上所有的虚拟机实例；
        ""
        agent_resources = self.manager.discover()
        with self.publish_context as publisher:
            cache = {}    
            """
            遍历任务（通道）；
            获取每个任务指定获取的监控项的采样数据；
            """
            for pollster in self.pollsters:
                key = pollster.name
                LOG.info(_("Polling pollster %s"), key)
                source_resources = list(self.resources[key].resources)
                try:
                    """
                    get_samples：获取nova管理下虚拟机实例的某一监控项的采样数据；
                    注：这些信息大部分是通过调用Hypervisor的API来获取的，Ceilometer提供了Libvirt的API和Hyper-V的API。
                    class CPUPollster(plugin.ComputePollster)----def get_samples(self, manager, cache, resources)
                    class CPUUtilPollster(plugin.ComputePollster)----def get_samples(self, manager, cache, resources)
                    class _Base(plugin.ComputePollster)----def get_samples(self, manager, cache, resources)
                    class _DiskRatesPollsterBase(plugin.ComputePollster)----def get_samples(self, manager, cache, resources)
                    class InstancePollster(plugin.ComputePollster)----def get_samples(manager, cache, resources)
                    class InstanceFlavorPollster(plugin.ComputePollster)----def get_samples(manager, cache, resources)
                    class MemoryUsagePollster(plugin.ComputePollster)----def get_samples(self, manager, cache, resources)
                    class _Base(plugin.ComputePollster)----def get_samples(self, manager, cache, resources)
                    """
                    samples = list(pollster.obj.get_samples(
                        self.manager,
                        cache,
                        resources=source_resources or agent_resources,
                    ))
                    
                    """
                    实现发布监控项采样数据样本；
                    """
                    publisher(samples)
                except Exception as err:
                    LOG.warning(_(
                        'Continue after error from %(name)s: %(error)s')
                        % ({'name': pollster.name, 'error': err}),
                        exc_info=True)

方法小结：
本方法周期性地执行以下操作：
（1）遍历任务（通道），获取每个任务指定获取的监控项的采样数据；
（2）针对每个监控项的采样数据，实现发布监控项采样数据样本到消息队列；

class PublishContext----def __enter__
在上述代码中，发布监控项采样样本数据是由这个方法实现的，这个方法主要实现了以下内容：实现发布监控项采样数据样本，其中实现采样数据发布的方式有三种，即RPC/UDP/FILE；
（1）class FilePublisher(publisher.PublisherBase)----def publish_samples(self, context, samples)；
实现发布采样数据到一个日志文件；
（2）class RPCPublisher(publisher.PublisherBase)----def publish_samples(self, context, samples)；
实现通过RPC发布采样数据；
* 从若干采样数据信息samples中获取提取数据形成信息格式meters，为信息的发布或存储做准备；
* 将之前从采样数据中提取的信息meters包装成msg；
* 将匹配的topic，msg添加到本地队列local_queue中，topic默认为metering；
* 实现发布本地队列local_queue中的所有数据信息到队列metering中；
* 其中，消息msg中封装的'method'方法为'record_metering_data'，即当消息被消费时，将会执行方法record_metering_data，实现存储到数据存储系统中（数据库）；
（3）class UDPPublisher(publisher.PublisherBase)----def publish_samples(self, context, samples)
通过UDP发布采样数据；

class PublishContext(object):  
        def __enter__(self):  
            """ 
            实现发布监控项采样数据样本； 
            publish_samples： 
            1.class FilePublisher(publisher.PublisherBase)----def publish_samples(self, context, samples)； 
            实现发布采样数据到一个日志文件； 
            2.class RPCPublisher(publisher.PublisherBase)----def publish_samples(self, context, samples)； 
            通过RPC发布采样数据； 
            * 从若干采样数据信息samples中获取提取数据形成信息格式meters，为信息的发布或存储做准备； 
            * 将之前从采样数据中提取的信息meters包装成msg； 
            * 将匹配的topic，msg添加到本地队列local_queue中，topic默认为metering； 
            * 实现发布本地队列local_queue中的所有数据信息到队列metering中； 
            * 其中，消息msg中封装的'method'方法为'record_metering_data'，即当消息被消费时，将会 
              执行方法record_metering_data，实现存储到数据存储系统中（数据库）； 
            3.class UDPPublisher(publisher.PublisherBase)----def publish_samples(self, context, samples) 
            通过UDP发布采样数据； 
            """  
            def p(samples):  
                for p in self.pipelines:  
                    p.publish_samples(self.context,  
                                      samples)  
            return p

[b]2.1 实现发布采样数据到一个日志文件[/b]

class FilePublisher(publisher.PublisherBase):  
        def publish_samples(self, context, samples):  
            if self.publisher_logger:  
                for sample in samples:  
                    self.publisher_logger.info(sample.as_dict())

2.2 通过RPC发布采样数据[b]（具体见代码注释）[/b]

class RPCPublisher(publisher.PublisherBase):  
        def publish_samples(self, context, samples):  
            """ 
            通过RPC发布信息； 
            1.从若干采样数据信息samples中获取提取数据形成信息格式meters，为信息的发布或存储做准备； 
            2.将之前从采样数据中提取的信息meters包装成msg； 
            3.将匹配的topic，msg添加到本地队列local_queue中，topic默认为metering； 
            4.实现发布本地队列local_queue中的所有数据信息到队列metering中； 
            5.其中，消息msg中封装的'method'方法为'record_metering_data'，即当消息被消费时，将会 
              执行方法record_metering_data，实现存储到数据存储系统中（数据库）； 
            """  
      
            # 从若干采样数据信息中获取提取数据形成信息格式，为信息的发布或存储做准备；  
            meters = [  
                # meter_message_from_counter：  
                # 为一个监控采样数据做好准备被发布或存储；  
                # 从一个采样数据信息中获取提取信息形成msg；  
                utils.meter_message_from_counter(  
                    sample,  
                    cfg.CONF.publisher.metering_secret)  
                for sample in samples  
            ]  
      
            # cfg.CONF.publisher_rpc.metering_topic：metering messages所使用的主题，默认为metering；  
            topic = cfg.CONF.publisher_rpc.metering_topic  
              
            # 将之前从采样数据中提取的信息meters包装成msg；  
            msg = {  
                'method': self.target,  
                'version': '1.0',  
                'args': {'data': meters},  
            }  
              
            # 将匹配的topic，msg添加到本地队列local_queue中，topic默认为metering；  
            self.local_queue.append((context, topic, msg))  
      
            if self.per_meter_topic:  
                for meter_name, meter_list in itertools.groupby(  
                        sorted(meters, key=operator.itemgetter('counter_name')),  
                        operator.itemgetter('counter_name')):  
                    msg = {  
                        'method': self.target,  
                        'version': '1.0',  
                        'args': {'data': list(meter_list)},  
                    }  
                    topic_name = topic + '.' + meter_name  
                    LOG.audit(_('Publishing %(m)d samples on %(n)s') % (  
                              {'m': len(msg['args']['data']), 'n': topic_name}))  
                    self.local_queue.append((context, topic_name, msg))  
      
            # 实现发布本地队列local_queue中的所有数据信息；  
            self.flush()

def flush(self):  
        """ 
        实现发布本地队列中的所有数据信息； 
        """          
        # 获取本地队列的数据信息；  
        queue = self.local_queue  
        self.local_queue = []  
              
        # 实现循环发布队列queue中的信息数据；  
        self.local_queue = self._process_queue(queue, self.policy) + \self.local_queue  
              
        if self.policy == 'queue':  
            self._check_queue_length()

@staticmethod  
    def _process_queue(queue, policy):  
        """ 
        实现循环发布队列queue中的信息数据； 
        """  
        while queue:  
            # 取出第一位的topic、msg等数据；  
            context, topic, msg = queue[0]  
            try:  
                # 实现远程发布信息，不返回任何值；  
                rpc.cast(context, topic, msg)  
            except (SystemExit, rpc.common.RPCException):  
                samples = sum([len(m['args']['data']) for n, n, m in queue])  
                if policy == 'queue':  
                    LOG.warn(_("Failed to publish %d samples, queue them"),samples)  
                    return queue  
                elif policy == 'drop':  
                    LOG.warn(_("Failed to publish %d samples, dropping them"),samples)  
                    return []  
                # default, occur only if rabbit_max_retries > 0  
                raise  
            else:  
                # 从队列中删除发布后的信息；  
                queue.pop(0)  
        return []

2.3 通过UDP发布采样数据[b]（具体见代码注释）[/b]

class UDPPublisher(publisher.PublisherBase):  
        def publish_samples(self, context, samples):  
            """ 
            通过UDP协议发送meter信息到服务器端，实现监控信息的发布； 
            """  
            for sample in samples:  
                """ 
                为一个监控采样数据做好准备被发布或存储； 
                从一个采样数据信息中获取提取信息形成msg； 
                """  
                msg = utils.meter_message_from_counter(  
                    sample,  
                    cfg.CONF.publisher.metering_secret)  
                host = self.host  
                port = self.port  
                LOG.debug(_("Publishing sample %(msg)s over UDP to "  
                            "%(host)s:%(port)d") % {'msg': msg, 'host': host,'port': port})  
                """ 
                通过UDP协议发送meter信息到服务器端，实现监控信息的发布； 
                """  
                try:  
                    self.socket.sendto(msgpack.dumps(msg),(self.host, self.port))  
                except Exception as e:  
                    LOG.warn(_("Unable to send sample over UDP"))  
                    LOG.exception(e)