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

问题导读
问题1:服务ceilometer-agent-compute的初始化操作实现了哪些内容的操作？
问题2:服务ceilometer-agent-compute的启动操作周期性地实现了哪些任务？








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()  
复制代码

1 服务ceilometer-agent-compute的初始化操作


服务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()  
复制代码

2 服务ceilometer-agent-compute的启动操作


这里服务ceilometer-agent-compute与服务ceilometer-agent-central的启动操作的实现代码是完全一致的，只不过所获取的监控任务不同而已，所调用的采集监控项采样数据的方法不同而已；


服务ceilometer-agent-compute的启动操作周期性地实现以下任务：


（1）遍历任务（通道），获取每个任务指定获取的监控项的采样数据；


（2）针对每个监控项的采样数据，实现发布监控项采样数据样本到消息队列，其中实现采样数据发布的方式有三种，即RPC/UDP/FILE；

    其中，RPC将会发布相关消息到消息队列，后续的collector组件服务将会监听相应的消息队列来获取这些数据信息；UDP将会建立socket建立一个信息通道，实现发送相关消息数据，而后续的collector组件服务将会通过这个信息通道接收相关的消息数据；FILE将会直接保存相关消息数据到指定的日志文件中。

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

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  
复制代码

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

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发布采样数据（具体见代码注释）

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发布采样数据（具体见代码注释）

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)  
复制代码

Ceilometer项目源码分析----ceilometer项目源码结构分析
Ceilometer项目源码分析----ceilometer报警器服务的实现概览
Ceilometer项目源码分析----ceilometer报警器状态评估方式
Ceilometer项目源码分析----ceilometer分布式报警系统的具体实现
Ceilometer项目源码分析----ceilometer-alarm-notifier服务的初始化和启动
Ceilometer项目源码分析----ceilometer-alarm-evaluator服务的初始化和启动
Ceilometer项目源码分析----ceilometer-agent-central服务的初始化和启动
Ceilometer项目源码分析----ceilometer-agent-compute服务的初始化和启动
Ceilometer项目源码分析----ceilometer-agent-notification服务的初始化和启动
Ceilometer项目源码分析----ceilometer-collector服务的初始化和启动