OpenStack监控架构解析

def main():
    conf = cfg.ConfigOpts()
    conf.register_cli_opts(CLI_OPTS)
    service.prepare_service(conf=conf)
    # cotyledon库是用来替代oslo.service库的，提供跟oslo.service类似的管理服务功能
    sm = cotyledon.ServiceManager()
    # 添加一个AgentManager管理类来运行
    # 初始化该实例会进行插件加载和消息发送实例初始化
    sm.add(create_polling_service, args=(conf,))
    oslo_config_glue.setup(sm, conf)
    # 调用AgentManager类实例的run方法启动服务
    sm.run()

class AgentManager(service_base.PipelineBasedService):

    def __init__(self, worker_id, conf, namespaces=None, pollster_list=None, ):

        # ceilometer-compute服务传入的是namespaces是compute
        namespaces = namespaces or [‘compute‘, ‘central‘]
        pollster_list = pollster_list or []
        ......
        # we‘ll have default [‘compute‘, ‘central‘] here if no namespaces will
        # be passed
        ‘‘‘ 加载采集插件
        # 根据namespance加载setup.cfg文件下的ceilometer.poll.compute、ceilometer.poll.central或
        ceilometer.builder.poll.central 这三个namespaces下对应的插件
        ‘‘‘
        extensions = (self._extensions(‘poll‘, namespace, self.conf).extensions
                      for namespace in namespaces)
        # get the extensions from pollster builder
        extensions_fb = (self._extensions_from_builder(‘poll‘, namespace)
                         for namespace in namespaces)
        if pollster_list:
            extensions = (moves.filter(_match, exts)
                          for exts in extensions)
            extensions_fb = (moves.filter(_match, exts)
                             for exts in extensions_fb)

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

        if self.extensions == []:
            raise EmptyPollstersList()

        # 加载ceilometer.discover.compute的插件，从setup.cfg中来看只有
        # ceilometer.compute.discovery:InstanceDiscovery用来发现宿主机下的云主机
        discoveries = (self._extensions(‘discover‘, namespace,
                                        self.conf).extensions
                       for namespace in namespaces)
        self.discoveries = list(itertools.chain(*list(discoveries)))
        self.polling_periodics = None

        # 获取一个工作负载分区协调类实例，用来协调多个采集程序worker时的分工处理
        self.partition_coordinator = coordination.PartitionCoordinator(
            self.conf)
        self.heartbeat_timer = utils.create_periodic(
            target=self.partition_coordinator.heartbeat,
            spacing=self.conf.coordination.heartbeat,
            run_immediately=True)

        # Compose coordination group prefix.
        # We‘ll use namespaces as the basement for this partitioning.
        namespace_prefix = ‘-‘.join(sorted(namespaces))
        self.group_prefix = (‘%s-%s‘ % (namespace_prefix, group_prefix)
                             if group_prefix else namespace_prefix)

        # 从该服务发送出去的消息都会携带ceilometer.polling这个publisher_id
        # notifier是oslo_messaging/notifier.py的Notifier类实例
        # 从配置文件中可以看到默认的driver是messagingv2，对应的是oslo_messageing/notify/messaging.py的MessagingV2Driver类实例
        # telemetry_driver = messagingv2
        self.notifier = oslo_messaging.Notifier(
            messaging.get_transport(self.conf),
            driver=self.conf.publisher_notifier.telemetry_driver,
            publisher_id="ceilometer.polling")

        self._keystone = None
        self._keystone_last_exception = None

def run(self):
        super(AgentManager, self).run()
        # 获取polling.yaml文件内容保存到列表中
        # 该文件中保存了需要监控的项目，默认设计*所有，但其实是可以自定义下，从而减少监控项，减少不必要的监控项
        # 调用ceilometer/pipeline.py的setup_polling函数
        self.polling_manager = pipeline.setup_polling(self.conf)
        self.join_partitioning_groups()
        # 启动polling采集任务
        self.start_polling_tasks()
        self.init_pipeline_refresh()

def start_polling_tasks(self):
    
    ......
    # 把要监测的资源全加载到polling_tasks中去
    # 返回的就是polling_tasks
    data = self.setup_polling_tasks()

    # Don‘t start useless threads if no task will run
    if not data:
        return

    # One thread per polling tasks is enough
    # 对于相同时间间隔的共用一个线程即可
    self.polling_periodics = periodics.PeriodicWorker.create(
        [], executor_factory=lambda:
        futures.ThreadPoolExecutor(max_workers=len(data)))

    # 对于每个时间间隔的polling_task都调用一次interval_task执行
    # interval_task又调用到了poll_and_notify方法
    for interval, polling_task in data.items():
        delay_time = (interval + delay_polling_time if delay_start
                      else delay_polling_time)

        @periodics.periodic(spacing=interval, run_immediately=False)
        def task(running_task):
            self.interval_task(running_task)

        # spawn_thread就是开一个线程，以daemon方式运行
        utils.spawn_thread(utils.delayed, delay_time,
                           self.polling_periodics.add, task, polling_task)

    utils.spawn_thread(self.polling_periodics.start, allow_empty=True)

# 将这个时间间隔的polling_task里的插件进行插件里的get_samples调用来采集数据并进行发送
def poll_and_notify(self):
    """Polling sample and notify."""
    .....
    # 对于每个source进行轮询
    for source_name in self.pollster_matches:
        # 对于每个插件进行轮询采集
        for pollster in self.pollster_matches[source_name]:
            key = Resources.key(source_name, pollster)
            candidate_res = list(
                self.resources[key].get(discovery_cache))
            if not candidate_res and pollster.obj.default_discovery:
                candidate_res = self.manager.discover(
                    [pollster.obj.default_discovery], discovery_cache)

            # 将要在该插件中采集的资源加入到polling_resources列表中
            # 比如对于cpu插件来说，这里的resources单元就是虚拟机
            polling_resources = []
            ......
            try:
                polling_timestamp = timeutils.utcnow().isoformat()
                # 调用插件的get_samples方法进行数据采集保存到samples变量中
                # 比如当前是采集cpu的插件，则是调用ceilometer.compute.pollsters.cpu:CPUPollster类的get_samples方法
                samples = pollster.obj.get_samples(
                    manager=self.manager,
                    cache=cache,
                    resources=polling_resources
                )
                sample_batch = []

                for sample in samples:
                    # Note(yuywz): Unify the timestamp of polled samples
                    sample.set_timestamp(polling_timestamp)
                    sample_dict = (
                        publisher_utils.meter_message_from_counter(
                            sample, self._telemetry_secret
                        ))
                    # 判断是否可以进行批量发送，默认是批量的，可以提高效率
                    if self._batch:
                        sample_batch.append(sample_dict)
                    else:
                        self._send_notification([sample_dict])

                if sample_batch:
                    # 批量发送采集到的数据
                    self._send_notification(sample_batch)

            except plugin_base.PollsterPermanentError as err:
                ......
            except Exception as err:
                ......

def _send_notification(self, samples):
    # <class ‘oslo_messaging.notify.notifier.Notifier‘>
    # ‘telemetry.polling‘是定义了event_type，消息事件类型
    self.manager.notifier.sample(
        {},
        ‘telemetry.polling‘,
        {‘samples‘: samples}
    )

def do_notify(ext):
    try:
        # oslo_messaging.notify.messaging.MessagingV2Driver
        ext.obj.notify(ctxt, msg, priority, retry or self.retry)
    except Exception as e:
        .....

def notify(self, ctxt, message, priority, retry):
    priority = priority.lower()
    for topic in self.topics:
        target = oslo_messaging.Target(topic=‘%s.%s‘ % (topic, priority))
        try:
            # 该transport是oslo_messaging/transport.pyTransport实例
            # 调用该实例的_send_notification方法其实又调用到它引用的driver的send_notification方法
            # 该driver根据配置文件定义是oslo_messaging/_drivers/impl_rabbit.py的RabbitDriver类实例
            # 调用了该父类的的send_notification方法进行消息发送
            self.transport._send_notification(target, ctxt, message,
                                              version=self.version,
                                              retry=retry)
        except Exception:
            .....

def get_samples(self, manager, cache, resources):
    for instance in resources:
        LOG.debug(‘checking instance %s‘, instance.id)
        try:
            # inspector是LibvirtInspector类实例
            # inspect_cpus方法其实是调用libvirt的接口去获取cpu的
            cpu_info = self.inspector.inspect_cpus(instance)
            LOG.debug("CPUTIME USAGE: %(instance)s %(time)d",
                      {‘instance‘: instance,
                       ‘time‘: cpu_info.time})
            cpu_num = {‘cpu_number‘: cpu_info.number}
            # 将cpu数据封装为一个sample类对象并返回
            # ceilometer/sample.py文件的sample类
            yield util.make_sample_from_instance(
                self.conf,
                instance,
                name=‘cpu‘,
                type=sample.TYPE_CUMULATIVE,
                unit=‘ns‘,
                volume=cpu_info.time,
                additional_metadata=cpu_num,
                monotonic_time=monotonic.monotonic()
            )
        except virt_inspector.InstanceNotFoundException as err:
            ......

def main():
    conf = service.prepare_service()

    sm = cotyledon.ServiceManager()
    # 初始化一个ceilometer/notification.py的NotificationService类实例
    sm.add(notification.NotificationService,
           workers=conf.notification.workers, args=(conf,))
    oslo_config_glue.setup(sm, conf)
    # 执行NotificationService类实例的run方法开始运行
    sm.run()

def run(self):
    ......
    self.pipeline_listener = None

    # pipeline、tranformer和publisher相关插件加载
    # 返回一个PipelineManager类实例
    self.pipeline_manager = pipeline.setup_pipeline(self.conf)

    # 跟setup_pipeline类似
    self.event_pipeline_manager = pipeline.setup_event_pipeline(self.conf)

    # 加载对应的driver实现，这里是oslo_message/_drivers/impl_rabbit.py的RabbitDriver类实现
    self.transport = messaging.get_transport(self.conf)

    # 如果是workload_partitioning设置为true，则需要初始化一个PartitionCoordinator类实例来协调
    if self.conf.notification.workload_partitioning:
        self.group_id = self.NOTIFICATION_NAMESPACE
        self.partition_coordinator = coordination.PartitionCoordinator(
            self.conf)
        # coordinator协调器
        self.partition_coordinator.start()
    else:
        ......

    # 该函数里判断了是否支持工作负载，如果是则返回SamplePipelineTransportManager类实例替换掉
    # PipelineManager类实例，否则还是PipelineManager类实例
    # 这两者的区别在于publisher函数实现是不一样的
    # SamplePipelineTransportManager的在调用publisher函数时会再发到消息队列中去保存，之后会再取出来处理再发到gnocchi-api上
    # PipelineManager的则直接去发到gnocchi-api服务上去了
    self.pipe_manager = self._get_pipe_manager(self.transport,
                                               self.pipeline_manager)
    self.event_pipe_manager = self._get_event_pipeline_manager(
        self.transport)

    # 获取ceilometer.notification的插件，并通过这些插件获取对应的target，再根据targets来找到要监听的
    # 队列在上面建立消费者获取消息并处理消息
    self._configure_main_queue_listeners(self.pipe_manager,
                                         self.event_pipe_manager)

    if self.conf.notification.workload_partitioning:
        .......

        # configure pipelines after all coordination is configured.
        with self.coord_lock:
            # 监听比如ceilometer-pipe-cpu_source:cpu_delta_sink-0.sample这样的队列的消费者
            # 因为配置了支持workload_partitioning后，经过转换的消息并没有直接发到gnocchi-api中去，而是
            # 分类保存到这些队列里了，再由这些队列接收到消息后再发送出去
            self._configure_pipeline_listener()

    self.init_pipeline_refresh()

def setup_pipeline(conf, transformer_manager=None):
    """Setup pipeline manager according to yaml config file."""
    # 加载setup.cfg文件中的ceilometer.transformer对应的模块
    default = extension.ExtensionManager(‘ceilometer.transformer‘)
    # 该配置文件对应的是/etc/ceilometer/pipeline.yaml
    cfg_file = conf.pipeline_cfg_file
    # SAMPLE_TYPE是一个字典，包含了SamplePipeline、SampleSource和SampleSink等类
    # 返回一个PipelineManager实例，该实例管理了SamplePipeline类实例数组，SamplePipeline又是管理了
    # SampleSource和SampleSink，用SampleSource中标志的sink来对应SampleSink中具体制定的transform和publisher
    return PipelineManager(conf, cfg_file, transformer_manager or default,
                           SAMPLE_TYPE)

class PipelineManager(ConfigManagerBase):
    def __init__(self, conf, cfg_file, transformer_manager,
                 p_type=SAMPLE_TYPE):
        ......
        # p_type[‘name‘]比如是sample
        # 这里只是获取一个PublisherManager类实例，还没有进行加载
        # 当需要获取时会通过调用该类实例的get方法动态获取对应的publisher
        publisher_manager = PublisherManager(self.conf, p_type[‘name‘])

        unique_names = set()
        sources = []
        # 解析sources，封装为SimpleSource
        for s in cfg.get(‘sources‘):
            name = s.get(‘name‘)
            if name in unique_names:
                raise PipelineException("Duplicated source names: %s" %
                                        name, self)
            else:
                unique_names.add(name)
                # p_type[‘source‘]对应的是SampleSource
                sources.append(p_type[‘source‘](s))
        unique_names.clear()

        # 解析sinks，封装为SimpleSink
        sinks = {}
        for s in cfg.get(‘sinks‘):
            name = s.get(‘name‘)
            if name in unique_names:
                raise PipelineException("Duplicated sink names: %s" %
                                        name, self)
            else:
                unique_names.add(name)
                # p_type[‘sink‘]对应的是SampleSink
                sinks[s[‘name‘]] = p_type[‘sink‘](self.conf, s,
                                                  transformer_manager,
                                                  publisher_manager)
        unique_names.clear()

        # 将加载的SampleSource和SampleSink封装成SamplePipeline
        for source in sources:
            source.check_sinks(sinks)
            for target in source.sinks:
                # p_type[‘pipeline‘]对应的是SamplePipeline
                pipe = p_type[‘pipeline‘](self.conf, source, sinks[target])
                if pipe.name in unique_names:
                    raise PipelineException(
                        "Duplicate pipeline name: %s. Ensure pipeline"
                        " names are unique. (name is the source and sink"
                        " names combined)" % pipe.name, cfg)
                else:
                    unique_names.add(pipe.name)
                    # 所以最后的结果都是保存到self.pipelines里
                    self.pipelines.append(pipe)
        unique_names.clear()

def _get_pipe_manager(self, transport, pipeline_manager):

    if self.conf.notification.workload_partitioning:
        pipe_manager = pipeline.SamplePipelineTransportManager(self.conf)
        for pipe in pipeline_manager.pipelines:
            key = pipeline.get_pipeline_grouping_key(pipe)
            pipe_manager.add_transporter(
                (pipe.source.support_meter, key or [‘resource_id‘],
                 self._get_notifiers(transport, pipe)))
    else:
        pipe_manager = pipeline_manager

    return pipe_manager

def publisher(self):
    return PublishContext(self.pipelines)
    
class PublishContext(object):
    .....
    def __enter__(self):
        def p(data):
            for p in self.pipelines:
                p.publish_data(data)
        return p
    .....

def publish_data(self, samples):
    .....
    supported = [s for s in samples if self.source.support_meter(s.name)
                 and self._validate_volume(s)]
    # 调用SinkSample类实例的publish_samples方法将监控数据通过_transform_sample处理后发送到gnocchi-api中去
    self.sink.publish_samples(supported)

def publisher(self):
    ......

    class PipelinePublishContext(object):
        def __enter__(self):
            def p(data):
                data = [data] if not isinstance(data, list) else data
                for datapoint in data:
                    serialized_data = serializer(datapoint)
                    for d_filter, grouping_keys, notifiers in transporters:
                        if d_filter(serialized_data[filter_attr]):
                            key = (hash_grouping(serialized_data,
                                                 grouping_keys)
                                   % len(notifiers))
                            notifier = notifiers[key]
                            # <class ‘oslo_messaging.notify.notifier.Notifier‘>
                            # 这里的event_type是ceilometer.pipeline，不是ceilometer.polling了
                            # 发送到类似ceilometer-pipe-xxx_source:xxx_sink-x.sample这样的队列去
                            notifier.sample({},
                                            event_type=event_type,
                                            payload=[serialized_data])
            return p

        def __exit__(self, exc_type, exc_value, traceback):
            pass

    return PipelinePublishContext()

def sample(self, messages):
    .....
    with self.publish_context as p:
        # 这里的p就又回到了ceilometer.pipeline.PipelineManager.publisher.PublishContext.p这个方法了
        # 将数据发送到gnocchi-api服务上去
        p(sorted(samples, key=methodcaller(‘get_iso_timestamp‘)))

def _configure_main_queue_listeners(self, pipe_manager,
                                    event_pipe_manager):
    # 获取ceilometer.notification命令空间里的插件
    notification_manager = self._get_notifications_manager(pipe_manager)


    ack_on_error = self.conf.notification.ack_on_event_error

    # 这里的endpoints是很关键的，因为当收到消息进行处理时会分派消息，分派时就是根据消息的类型分派到指定的插件类来处理消息
    endpoints = []
    endpoints.append(
        event_endpoint.EventsNotificationEndpoint(event_pipe_manager))

    targets = []
    for ext in notification_manager:
        handler = ext.obj
        # 针对每一个加载的插件获取它们的targets
        # 其中_sample的target为1个，<Target exchange=ceilometer, topic=notifications>
        # 比如http.request的则有多个，比如<Target exchange=nova, topic=notifications>，<Target exchange=glance, topic=notifications>
        for new_tar in handler.get_targets(self.conf):
            if new_tar not in targets:
                targets.append(new_tar)
        # 每个插件的对象实例保存到endpoints
        endpoints.append(handler)

    urls = self.conf.notification.messaging_urls or [None]
    for url in urls:
        transport = messaging.get_transport(self.conf, url)
        # 初始化一个solo_messaging/notify/listener.py文件的BatchNotificationServer类实例
        listener = messaging.get_batch_notification_listener(
            transport, targets, endpoints)
        # 启动监听，创建消费者，绑定收到消息时的回调函数
        listener.start()
        self.listeners.append(listener)

def get_batch_notification_listener(transport, targets, endpoints,
                                    executor=‘blocking‘, serializer=None,
                                    allow_requeue=False, pool=None,
                                    batch_size=None, batch_timeout=None):
    # 初始化一个oslo_messaging/notify/dispatcher.py的BatchNotificationDispatcher类实例
    dispatcher = notify_dispatcher.BatchNotificationDispatcher(
        endpoints, serializer)
    return BatchNotificationServer(
        transport, targets, dispatcher, executor, allow_requeue, pool,
        batch_size, batch_timeout
)

# 会被start方法调用
def _create_listener(self):
    return self.transport._listen_for_notifications(
        self._targets_priorities, self._pool, self._batch_size,
        self._batch_timeout
    )

def listen_for_notifications(self, targets_and_priorities, pool,
                             batch_size, batch_timeout):
    conn = self._get_connection(rpc_common.PURPOSE_LISTEN)
    conn.connection.rabbit_qos_prefetch_count = batch_size

    listener = AMQPListener(self, conn)
    # 根据每个target创建对应队列上的消费者
    for target, priority in targets_and_priorities:
        conn.declare_topic_consumer(
            exchange_name=self._get_exchange(target),
            topic=‘%s.%s‘ % (target.topic, priority),
            callback=listener, queue_name=pool)
    return base.PollStyleListenerAdapter(listener, batch_size,
                                         batch_timeout)

def _process_incoming(self, incoming):
    message = incoming[0]
    try:
        res = self.dispatcher.dispatch(message)
    except Exception:
        ......

class NotificationDispatcher(dispatcher.DispatcherBase):
    def __init__(self, endpoints, serializer):

        # endpoints：Notification插件集合
        # 对应ceilometer.notification加载的插件，该插件会接受它对应的消息并进行处理且最后调用publisher将处理后的消息发送出去
        self.endpoints = endpoints
        self.serializer = serializer or msg_serializer.NoOpSerializer()

        self._callbacks_by_priority = {}
        for endpoint, prio in itertools.product(endpoints, PRIORITIES):
            if hasattr(endpoint, prio):
                # 这里的方法就是sample、info、debug等，在ceilometer/agent/plugin_base.py中的NotificationBase类都会对应名字的方法实现
                # 注意了如果是ceilometer-pipe-cpu_source:cpu_delta_sink-0.sample类似这些队列来的则处理函数就不是sample、info、debug那些了
                # 而是SamplePipelineEndpoint类的sample方法了，因为endpoints不一样了
                method = getattr(endpoint, prio)
                screen = getattr(endpoint, ‘filter_rule‘, None)
                self._callbacks_by_priority.setdefault(prio, []).append(
                    (screen, method))

def dispatch(self, incoming):
    """Dispatch notification messages to the appropriate endpoint method.
    """
    # 将资源分配到合理的处理插件上去处理消息
    priority, raw_message, message = self._extract_user_message(incoming)

    if priority not in PRIORITIES:
        LOG.warning(_LW(‘Unknown priority "%s"‘), priority)
        return

    for screen, callback in self._callbacks_by_priority.get(priority,
                                                            []):
        if screen and not screen.match(message["ctxt"],
                                       message["publisher_id"],
                                       message["event_type"],
                                       message["metadata"],
                                       message["payload"]):
            continue

        ret = self._exec_callback(callback, message)
        if ret == NotificationResult.REQUEUE:
            return ret
    return NotificationResult.HANDLED

def sample(self, notifications):
    self._process_notifications(‘sample‘, notifications)

_process_notifications方法又调用了to_samples_and_publish方法：
File：ceilometer/agent/plugin_base.py:NotificationBase.to_samples_and_publish
def to_samples_and_publish(self, notification):
    with self.manager.publisher() as p:
        # 这里处理好的sample的格式类似于
        # <name: bandwidth, volume: 252, resource_id: a16d3949-29a6-4d88-b9a1-c7fdc97d0e4f, timestamp: 2019-07-04 03:02:25.181893>
        # p是ceilometer.pipeline.SamplePipelineTransportManager.publisher.PipelinePublishContext.p方法，这个会再发到消息队列上去
        # 如类似ceilometer-pipe-cpu_source:cpu_delta_sink-0.sample这种队列
        # 如果workload_partitioning设置为false则调用的是ceilometer.pipeline.PipelineManager.publisher.PublishContext.p方法，这个是
        # 直接发送gnocchi-api上去的
        # 这个process_notification是调用对应的插件的process_notification方法对监控数据进行处理
        p(list(self.process_notification(notification)))

def p(data):
    for p in self.pipelines:
        p.publish_data(data)

def publish_data(self, samples):
    ......
    # 调用SampleSink类实例的publish_samples方法将监控数据通过_transform_sample处理后发送到gnocchi-api中去
    self.sink.publish_samples(supported)

def _publish_samples(self, start, samples):

    transformed_samples = []
    if not self.transformers:
        transformed_samples = samples
    else:
        for sample in samples:
            sample = self._transform_sample(start, sample)
            if sample:
                transformed_samples.append(sample)

    if transformed_samples:
        # 配置文件配的是gnocchi，根据setup.cfg中的指定
        # gnocchi = ceilometer.publisher.direct:DirectPublisher
        # 所以这里其实是调用到ceilometer/publisher/direct.py文件的DirectPublisher类实例的publish_samples方法
        for p in self.publishers:
            try:
                # 所以这里其实是调用到ceilometer/publisher/direct.py文件的DirectPublisher类实例的publish_samples方法
                p.publish_samples(transformed_samples)
            except Exception:
                .....

def publish_samples(self, samples):
    .....
    # 这里是调用到ceilometer/dipatcher/gnocchi.py里的record_metering_data
    self.get_sample_dispatcher().record_metering_data([
        utils.meter_message_from_counter(sample, secret=None)
        for sample in samples])

# 将sample进行加工后发送到gnocchi中去
def record_metering_data(self, data):
    # 处理data
    .......

    try:
        # 连接gnocchiclient发送数据到gnocchi中去保存数据
        self.batch_measures(measures, gnocchi_data, stats)
    except (gnocchi_exc.ClientException,
            ka_exceptions.ConnectFailure) as e:
        ......

def batch_measures(self, measures, resource_infos, stats):
    try:
        # 调用gnocchiclient/v1/metric.py的batch_resources_metrics_measures方法发起http的post请求
        self._gnocchi.metric.batch_resources_metrics_measures(
            measures, create_metrics=True)
    except gnocchi_exc.BadRequest as e:
        ......

from gnocchi.rest import app
application = app.build_wsgi_app()

def load_app(conf, indexer=None, storage=None,
             not_implemented_middleware=True):
    global APPCONFIGS

    # NOTE(sileht): We load config, storage and indexer,
    # so all
    if not storage:
        storage = gnocchi_storage.get_driver(conf)
    if not indexer:
        indexer = gnocchi_indexer.get_driver(conf)
        indexer.connect()

    # Build the WSGI app
    cfg_path = conf.api.paste_config
    if not os.path.isabs(cfg_path):
        cfg_path = conf.find_file(cfg_path)

    if cfg_path is None or not os.path.exists(cfg_path):
        raise cfg.ConfigFilesNotFoundError([conf.api.paste_config])

    config = dict(conf=conf, indexer=indexer, storage=storage,
                  not_implemented_middleware=not_implemented_middleware)
    configkey = str(uuid.uuid4())
    APPCONFIGS[configkey] = config

    LOG.info("WSGI config used: %s", cfg_path)
    appname = "gnocchi+" + conf.api.auth_mode
    app = deploy.loadapp("config:" + cfg_path, name=appname,
                         global_conf={‘configkey‘: configkey})
return cors.CORS(app, conf=conf)

def batch_resources_metrics_measures(self, measures, create_metrics=False):

    # v1/batch/resources/metrics/measures
    # 对应gnocchi服务的pecan url路径查找可知识调用了gnocchi/rest/__init__.py的ResourcesMetricsMeasuresBatchController类的post方法
    return self._post(
        self.resources_batch_url,
        headers={‘Content-Type‘: "application/json"},
        data=jsonutils.dumps(measures),
        params=dict(create_metrics=create_metrics))

class ResourcesMetricsMeasuresBatchController(rest.RestController):
    @pecan.expose(‘json‘)
    def post(self, create_metrics=False):
        # 判断该监控名字是否已存在和封装数据等逻辑
        ......

        # 保存监控数据，等待gnocchi-metricd服务对它异步处理
        # pecan.request.storage这个对象是<class ‘gnocchi.storage.ceph.CephStorage‘>
        # pecan.request.storage.incoming这个是<class ‘gnocchi.storage.incoming.file.FileStorage‘>
        # 这个应该就是我们配置文件中配置的临时的先放file里（可查看配置文件的[incoming]项，存到/var/lib/gnocchi/该目录下），最终的聚合计算后的数据放ceph里
        pecan.request.storage.incoming.add_measures_batch(
            dict((metric,
                 body_by_rid[metric.resource_id][metric.name])
                 for metric in known_metrics))

        ....

def _store_new_measures(self, metric, data):
    tmpfile = tempfile.NamedTemporaryFile(
        prefix=‘gnocchi‘, dir=self.basepath_tmp,
        delete=False)
    tmpfile.write(data)
    tmpfile.close()
    # 这个目录下/var/lib/gnocchi/
    path = self._build_measure_path(metric.id, True)
    while True:
        try:
            os.rename(tmpfile.name, path)
            break
        except OSError as e:
            ......

# 程序启动时入口点
def metricd():
    conf = cfg.ConfigOpts()
    conf.register_cli_opts([
        cfg.IntOpt("stop-after-processing-metrics",
                   default=0,
                   min=0,
                   help="Number of metrics to process without workers, "
                   "for testing purpose"),
    ])
    conf = service.prepare_service(conf=conf)

    if conf.stop_after_processing_metrics:
        metricd_tester(conf)
    else:
        MetricdServiceManager(conf).run()

class MetricdServiceManager(cotyledon.ServiceManager):
    def __init__(self, conf):
        super(MetricdServiceManager, self).__init__()
        oslo_config_glue.setup(self, conf)

        self.conf = conf
        self.queue = multiprocessing.Manager().Queue()

        # 从measure storage中取出监控数据存放到queue队列中
        self.add(MetricScheduler, args=(self.conf, self.queue))
        # 不断的从queue队列中获取数据并进行聚合计算并保存到aggregate storage中去
        self.metric_processor_id = self.add(
            MetricProcessor, args=(self.conf, self.queue),
            workers=conf.metricd.workers)
        if self.conf.metricd.metric_reporting_delay >= 0:
            self.add(MetricReporting, args=(self.conf,))
        # 用来删除在index数据库中资源标志为delete的监控数据
        self.add(MetricJanitor, args=(self.conf,))

        self.register_hooks(on_reload=self.on_reload)

def run(self):
    # run方法会一直运行
    super(MetricdServiceManager, self).run()
    # 在程序结束或停止时关闭queue
    self.queue.close()