gnocchi-采样数据存储流程分析(002)--数据的异步统计
1 数据的异步统计
在gnocchi的核心思想中，是通过后台的异步处理ceilometer发送过来的采样数据，然后根据存储策略定义的汇聚方式，对数据进行预处理。然后用户获取统计数据的时候，直接获取到对应的已经统计好的数据，以此来提升性能，以及减少存储的采样数据。这边主要分析下gnocchi的异步统计流程。
进程为：
/usr/bin/python2 /usr/bin/gnocchi-metricd --logfile /var/log/gnocchi/metricd.log
启动入口：
1.1 start
源码位于： gnocchi/cli.py 中的metricd中，其异步处理对象为MetricProcessor：
def metricd():
conf = service.prepare_service()
if (conf.storage.metric_reporting_delay <
conf.storage.metric_processing_delay):
LOG.error("Metric reporting must run less frequently then processing")
sys.exit(0)
signal.signal(signal.SIGTERM, _metricd_terminate)
try:
queues = []
workers = []
for worker in range(conf.metricd.workers):
queue = multiprocessing.Queue()
metric_worker = MetricProcessor(
conf, worker, conf.storage.metric_processing_delay, queue)
metric_worker.start()
queues.append(queue)
workers.append(metric_worker)
metric_report = MetricReporting(
conf, 0, conf.storage.metric_reporting_delay, queues)
metric_report.start()
workers.append(metric_report)
for worker in workers:
worker.join()
except KeyboardInterrupt:
_metricd_cleanup(workers)
sys.exit(0)
except Exception:
LOG.warning("exiting", exc_info=True)
_metricd_cleanup(workers)
sys.exit(1)
1.2 MetricProcessor
该类主要是调用存储后端的 process_background_tasks
class MetricProcessor(MetricProcessBase):
def __init__(self, conf, worker_id=0, interval_delay=0, queue=None):
super(MetricProcessor, self).__init__(conf, worker_id, interval_delay)
self.queue = queue
self.block_size = 128
def _run_job(self):
try:
if self.queue:
while not self.queue.empty():
self.block_size = self.queue.get()
LOG.debug("Re-configuring worker to handle up to %s "
"metrics", self.block_size)
self.store.process_background_tasks(self.index, self.block_size)
except Exception:
LOG.error("Unexpected error during measures processing",
exc_info=True)
1.3 process_background_tasks
process_background_tasks代码定义于存储后端的父类驱动中：
Gnocchi/storage/__init__.py中的StorageDriver中的process_background_tasks
def process_background_tasks(self, index, block_size=128, sync=False):
"""Process background tasks for this storage.
This calls :func:`process_measures` to process new measures and
:func:`expunge_metrics` to expunge deleted metrics.
:param index: An indexer to be used for querying metrics
:param block_size: number of metrics to process
:param sync: If True, then process everything synchronously and raise
on error
:type sync: bool
"""
LOG.debug("Processing new and to delete measures")
try:
self.process_measures(index, block_size, sync)
except Exception:
if sync:
raise
LOG.error("Unexpected error during measures processing",
exc_info=True)
LOG.debug("Expunging deleted metrics")
try:
#这个主要是删除被删除的metric信息，并将相关采集数据删除
self.expunge_metrics(index, sync)
except Exception:
if sync:
raise
LOG.error("Unexpected error during deleting metrics",
exc_info=True)
1.4 process_measures
该方法定义于gnocchi/storage/_carbonara.py中
def process_measures(self, indexer, block_size, sync=False):
#获取当前上报的采样metric记录
metrics_to_process = self._list_metric_with_measures_to_process(
block_size, full=sync)
metrics = indexer.list_metrics(ids=metrics_to_process)
# This build the list of deleted metrics, i.e. the metrics we have
# measures to process for but that are not in the indexer anymore.
deleted_metrics_id = (set(map(uuid.UUID, metrics_to_process))
- set(m.id for m in metrics))
for metric_id in deleted_metrics_id:
# NOTE(jd): We need to lock the metric otherwise we might delete
# measures that another worker might be processing. Deleting
# measurement files under its feet is not nice!
with self._lock(metric_id)(blocking=sync):
#删除未处理的并上报的采样数据，主要是metric统计和采样的上报的异步的，存在删除的时候，但是采样值还在没被统计的情况。
self._delete_unprocessed_measures_for_metric_id(metric_id)
for metric in metrics:
lock = self._lock(metric.id)
agg_methods = list(metric.archive_policy.aggregation_methods)
# Do not block if we cannot acquire the lock, that means some other
# worker is doing the job. We'll just ignore this metric and may
# get back later to it if needed.
if lock.acquire(blocking=sync):
try:
LOG.debug("Processing measures for %s" % metric)
# measures 为新增加的采样的数据
#存放的路径为/var/lib/gnocchi/measure/$metric.id
with self._process_measure_for_metric(metric) as measures:
# NOTE(mnaser): The metric could have been handled by
# another worker, ignore if no measures.
if len(measures) == 0:
LOG.debug("Skipping %s (already processed)"
% metric)
continue
try:
with timeutils.StopWatch() as sw:
#file存储路径为/var/lib/gnocchi/$metric.id/none
raw_measures = (
self._get_unaggregated_timeserie(
metric)
)
LOG.debug(
"Retrieve unaggregated measures "
"for %s in %.2fs"
% (metric.id, sw.elapsed()))
except storage.MetricDoesNotExist:
try:
self._create_metric(metric)
except storage.MetricAlreadyExists:
# Created in the mean time, do not worry
pass
ts = None
else:
try:
ts = carbonara.BoundTimeSerie.unserialize(
raw_measures)
except ValueError:
ts = None
LOG.error(
"Data corruption detected for %s "
"unaggregated timeserie, "
"recreating an empty one."
% metric.id)
if ts is None:
# This is the first time we treat measures for this
# metric, or data are corrupted, create a new one
mbs = metric.archive_policy.max_block_size
ts = carbonara.BoundTimeSerie(
block_size=mbs,
back_window=metric.archive_policy.back_window)
def _map_add_measures(bound_timeserie):
self._map_in_thread(
self._add_measures,
((aggregation, d, metric, bound_timeserie)
for aggregation in agg_methods
for d in metric.archive_policy.definition))
with timeutils.StopWatch() as sw:
ts.set_values(
measures,
before_truncate_callback=_map_add_measures,
ignore_too_old_timestamps=True)
LOG.debug(
"Computed new metric %s with %d new measures "
"in %.2f seconds"
% (metric.id, len(measures), sw.elapsed()))
self._store_unaggregated_timeserie(metric,
ts.serialize())
except Exception:
if sync:
raise
LOG.error("Error processing new measures", exc_info=True)
finally:
lock.release()
1.4.1 _list_metric_with_measures_to_process
该方法定义在对应的存储后端代码中，本文以file为例：
Gnocchi/gnocchi/storage/file.py 中的，
#获取当前上报的采样metric记录
def _list_metric_with_measures_to_process(self, block_size, full=False):
if full:
return os.listdir(self.measure_path)
return os.listdir(self.measure_path)[
block_size * self.partition:block_size * (self.partition + 1)]
1.4.2 _delete_unprocessed_measures_for_metric_id
该源码位于Gnocchi/gnocchi/storage/file.py 中的，
#删除未处理的并上报的采样数据
def _delete_unprocessed_measures_for_metric_id(self, metric_id):
files = self._list_measures_container_for_metric_id(metric_id)
self._delete_measures_files_for_metric_id(metric_id, files)
1.4.3 _process_measure_for_metric
该源码位于Gnocchi/gnocchi/storage/file.py 中的，
#获取保存在measure/metric_id中的采样信息，处理后删除
@contextlib.contextmanager
def _process_measure_for_metric(self, metric):
files = self._list_measures_container_for_metric_id(metric.id)
measures = []
for f in files:
abspath = self._build_measure_path(metric.id, f)
with open(abspath, "rb") as e:
measures.extend(self._unserialize_measures(e.read()))
yield measures
self._delete_measures_files_for_metric_id(metric.id, files)
1.4.4 _get_unaggregated_timeserie
该源码位于Gnocchi/gnocchi/storage/file.py 中的，
#获取保存的未统计过的采样数据
#file存储路径为/var/lib/gnocchi/$metric.id/none
def _get_unaggregated_timeserie(self, metric):
path = self._build_unaggregated_timeserie_path(metric)
try:
with open(path, 'rb') as f:
return f.read()
except IOError as e:
if e.errno == errno.ENOENT:
raise storage.MetricDoesNotExist(metric)
raise
def _get_unaggregated_timeserie(self, metric):
path = self._build_unaggregated_timeserie_path(metric)
try:
with open(path, 'rb') as f:
return f.read()
except IOError as e:
if e.errno == errno.ENOENT:
raise storage.MetricDoesNotExist(metric)
raise
def _build_unaggregated_timeserie_path(self, metric):
return os.path.join(self._build_metric_dir(metric), 'none')
def _store_unaggregated_timeserie(self, metric, data):
self._atomic_file_store(
self._build_unaggregated_timeserie_path(metric),
data)
1.4.5 _create_metric
#根据metric生成对应agg数据的存储路径：
def _create_metric(self, metric):
path = self._build_metric_dir(metric)
try:
os.mkdir(path, 0o750)
except OSError as e:
if e.errno == errno.EEXIST:
raise storage.MetricAlreadyExists(metric)
raise
for agg in metric.archive_policy.aggregation_methods:
try:
os.mkdir(self._build_metric_path(metric, agg), 0o750)
except OSError as e:
if e.errno != errno.EEXIST:
raise
1.4.6 set_values
进行数据的汇聚处理
该源码位于gnocchi/carbonara.py中的 set_values
# 调用panna库，进行数据的异步处理功能
def set_values(self, values, before_truncate_callback=None,
ignore_too_old_timestamps=False):
if self.block_size is not None and not self.ts.empty:
values = sorted(values, key=operator.itemgetter(0))
first_block_timestamp = self._first_block_timestamp()
if ignore_too_old_timestamps:
for index, (timestamp, value) in enumerate(values):
if timestamp >= first_block_timestamp:
values = values[index:]
break
else:
values = []
else:
# Check that the smallest timestamp does not go too much back
# in time.
smallest_timestamp = values[0][0]
if smallest_timestamp < first_block_timestamp:
raise NoDeloreanAvailable(first_block_timestamp,
smallest_timestamp)
super(BoundTimeSerie, self).set_values(values)
if before_truncate_callback:
before_truncate_callback(self)
self._truncate()
1.4.7 _add_measures
代码位于 gnocchi/storage/_carbonara.py中
#这里就是进行最终的异步统计过程，使用第三方统计工具pandas，并根据archive policy定义的时间戳，保留最长的记录。
#并删除超过时间戳的数据
def _map_add_measures(bound_timeserie):
self._map_in_thread(
self._add_measures,
((aggregation, d, metric, bound_timeserie)
for aggregation in agg_methods
for d in metric.archive_policy.definition))
def _add_measures(self, aggregation, archive_policy_def,
metric, timeserie):
with timeutils.StopWatch() as sw:
ts = self._get_measures_timeserie(metric, aggregation,
archive_policy_def.granularity,
timeserie.first, timeserie.last)
LOG.debug("Retrieve measures"
"for %s/%s/%s in %.2fs"
% (metric.id, aggregation, archive_policy_def.
granularity, sw.elapsed()))
ts.update(timeserie)
with timeutils.StopWatch() as sw:
for key, split in ts.split():
self._store_metric_measures(metric, key, aggregation,
archive_policy_def.granularity,
split.serialize())
LOG.debug("Store measures for %s/%s/%s in %.2fs"
% (metric.id, aggregation,
archive_policy_def.granularity, sw.elapsed()))
if ts.last and archive_policy_def.timespan:
with timeutils.StopWatch() as sw:
oldest_point_to_keep = ts.last - datetime.timedelta(
seconds=archive_policy_def.timespan)
self._delete_metric_measures_before(
metric, aggregation, archive_policy_def.granularity,
oldest_point_to_keep)
LOG.debug("Expire measures for %s/%s/%s in %.2fs"
% (metric.id, aggregation,
archive_policy_def.granularity, sw.elapsed()))