From ad476e5611075985d269c253d65525552e353beb Mon Sep 17 00:00:00 2001 From: Aarni Koskela Date: Mon, 21 Jan 2019 14:39:25 +0200 Subject: [PATCH 1/6] Packagify multiprocess bits Signed-off-by: Aarni Koskela --- .../__init__.py} | 8 +- .../{ => multiprocess}/mmap_dict.py | 0 .../multiprocess/mmaped_value.py | 73 +++++++++++++++++++ prometheus_client/values.py | 73 +------------------ tests/test_multiprocess.py | 33 +++++---- 5 files changed, 95 insertions(+), 92 deletions(-) rename prometheus_client/{multiprocess.py => multiprocess/__init__.py} (97%) rename prometheus_client/{ => multiprocess}/mmap_dict.py (100%) create mode 100644 prometheus_client/multiprocess/mmaped_value.py diff --git a/prometheus_client/multiprocess.py b/prometheus_client/multiprocess/__init__.py similarity index 97% rename from prometheus_client/multiprocess.py rename to prometheus_client/multiprocess/__init__.py index 30ed312c..449de748 100644 --- a/prometheus_client/multiprocess.py +++ b/prometheus_client/multiprocess/__init__.py @@ -1,5 +1,3 @@ -#!/usr/bin/python - from __future__ import unicode_literals from collections import defaultdict @@ -7,10 +5,10 @@ import json import os -from .metrics_core import Metric +from ..metrics_core import Metric from .mmap_dict import MmapedDict -from .samples import Sample -from .utils import floatToGoString +from ..samples import Sample +from ..utils import floatToGoString class MultiProcessCollector(object): diff --git a/prometheus_client/mmap_dict.py b/prometheus_client/multiprocess/mmap_dict.py similarity index 100% rename from prometheus_client/mmap_dict.py rename to prometheus_client/multiprocess/mmap_dict.py diff --git a/prometheus_client/multiprocess/mmaped_value.py b/prometheus_client/multiprocess/mmaped_value.py new file mode 100644 index 00000000..e74da31f --- /dev/null +++ b/prometheus_client/multiprocess/mmaped_value.py @@ -0,0 +1,73 @@ +import os +from threading import Lock + +from prometheus_client.multiprocess import MmapedDict +from prometheus_client.multiprocess.mmap_dict import mmap_key + + +def MmapedValue(_pidFunc=os.getpid): + files = {} + values = [] + pid = {'value': _pidFunc()} + # Use a single global lock when in multi-processing mode + # as we presume this means there is no threading going on. + # This avoids the need to also have mutexes in __MmapDict. + lock = Lock() + + class MmapedValue(object): + '''A float protected by a mutex backed by a per-process mmaped file.''' + + _multiprocess = True + + def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', **kwargs): + self._params = typ, metric_name, name, labelnames, labelvalues, multiprocess_mode + with lock: + self.__check_for_pid_change() + self.__reset() + values.append(self) + + def __reset(self): + typ, metric_name, name, labelnames, labelvalues, multiprocess_mode = self._params + if typ == 'gauge': + file_prefix = typ + '_' + multiprocess_mode + else: + file_prefix = typ + if file_prefix not in files: + filename = os.path.join( + os.environ['prometheus_multiproc_dir'], + '{0}_{1}.db'.format(file_prefix, pid['value'])) + + files[file_prefix] = MmapedDict(filename) + self._file = files[file_prefix] + self._key = mmap_key(metric_name, name, labelnames, labelvalues) + self._value = self._file.read_value(self._key) + + def __check_for_pid_change(self): + actual_pid = _pidFunc() + if pid['value'] != actual_pid: + pid['value'] = actual_pid + # There has been a fork(), reset all the values. + for f in files.values(): + f.close() + files.clear() + for value in values: + value.__reset() + + def inc(self, amount): + with lock: + self.__check_for_pid_change() + self._value += amount + self._file.write_value(self._key, self._value) + + def set(self, value): + with lock: + self.__check_for_pid_change() + self._value = value + self._file.write_value(self._key, self._value) + + def get(self): + with lock: + self.__check_for_pid_change() + return self._value + + return MmapedValue diff --git a/prometheus_client/values.py b/prometheus_client/values.py index c9eed177..7a63c70a 100644 --- a/prometheus_client/values.py +++ b/prometheus_client/values.py @@ -3,8 +3,6 @@ import os from threading import Lock -from .mmap_dict import mmap_key, MmapedDict - class MutexValue(object): '''A float protected by a mutex.''' @@ -28,81 +26,14 @@ def get(self): return self._value -def MultiProcessValue(_pidFunc=os.getpid): - files = {} - values = [] - pid = {'value': _pidFunc()} - # Use a single global lock when in multi-processing mode - # as we presume this means there is no threading going on. - # This avoids the need to also have mutexes in __MmapDict. - lock = Lock() - - class MmapedValue(object): - '''A float protected by a mutex backed by a per-process mmaped file.''' - - _multiprocess = True - - def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', **kwargs): - self._params = typ, metric_name, name, labelnames, labelvalues, multiprocess_mode - with lock: - self.__check_for_pid_change() - self.__reset() - values.append(self) - - def __reset(self): - typ, metric_name, name, labelnames, labelvalues, multiprocess_mode = self._params - if typ == 'gauge': - file_prefix = typ + '_' + multiprocess_mode - else: - file_prefix = typ - if file_prefix not in files: - filename = os.path.join( - os.environ['prometheus_multiproc_dir'], - '{0}_{1}.db'.format(file_prefix, pid['value'])) - - files[file_prefix] = MmapedDict(filename) - self._file = files[file_prefix] - self._key = mmap_key(metric_name, name, labelnames, labelvalues) - self._value = self._file.read_value(self._key) - - def __check_for_pid_change(self): - actual_pid = _pidFunc() - if pid['value'] != actual_pid: - pid['value'] = actual_pid - # There has been a fork(), reset all the values. - for f in files.values(): - f.close() - files.clear() - for value in values: - value.__reset() - - def inc(self, amount): - with lock: - self.__check_for_pid_change() - self._value += amount - self._file.write_value(self._key, self._value) - - def set(self, value): - with lock: - self.__check_for_pid_change() - self._value = value - self._file.write_value(self._key, self._value) - - def get(self): - with lock: - self.__check_for_pid_change() - return self._value - - return MmapedValue - - def get_value_class(): # Should we enable multi-process mode? # This needs to be chosen before the first metric is constructed, # and as that may be in some arbitrary library the user/admin has # no control over we use an environment variable. if 'prometheus_multiproc_dir' in os.environ: - return MultiProcessValue() + from prometheus_client.multiprocess.mmaped_value import MmapedValue + return MmapedValue() else: return MutexValue diff --git a/tests/test_multiprocess.py b/tests/test_multiprocess.py index d57b23a0..d9f0e699 100644 --- a/tests/test_multiprocess.py +++ b/tests/test_multiprocess.py @@ -6,14 +6,15 @@ import sys import tempfile -from prometheus_client import core, mmap_dict, values +from prometheus_client import core, values from prometheus_client.core import ( CollectorRegistry, Counter, Gauge, Histogram, Sample, Summary, ) from prometheus_client.multiprocess import ( mark_process_dead, MultiProcessCollector, -) -from prometheus_client.values import MultiProcessValue, MutexValue + mmap_dict) +from prometheus_client.values import MutexValue +from prometheus_client.multiprocess.mmaped_value import MmapedValue if sys.version_info < (2, 7): # We need the skip decorators from unittest2 on Python 2.6. @@ -28,7 +29,7 @@ class TestMultiProcess(unittest.TestCase): def setUp(self): self.tempdir = tempfile.mkdtemp() os.environ['prometheus_multiproc_dir'] = self.tempdir - values.ValueClass = MultiProcessValue(lambda: 123) + values.ValueClass = MmapedValue(lambda: 123) self.registry = CollectorRegistry() self.collector = MultiProcessCollector(self.registry, self.tempdir) @@ -43,7 +44,7 @@ def tearDown(self): def test_counter_adds(self): c1 = Counter('c', 'help', registry=None) - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) c2 = Counter('c', 'help', registry=None) self.assertEqual(0, self.registry.get_sample_value('c_total')) c1.inc(1) @@ -52,7 +53,7 @@ def test_counter_adds(self): def test_summary_adds(self): s1 = Summary('s', 'help', registry=None) - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) s2 = Summary('s', 'help', registry=None) self.assertEqual(0, self.registry.get_sample_value('s_count')) self.assertEqual(0, self.registry.get_sample_value('s_sum')) @@ -63,7 +64,7 @@ def test_summary_adds(self): def test_histogram_adds(self): h1 = Histogram('h', 'help', registry=None) - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) h2 = Histogram('h', 'help', registry=None) self.assertEqual(0, self.registry.get_sample_value('h_count')) self.assertEqual(0, self.registry.get_sample_value('h_sum')) @@ -76,7 +77,7 @@ def test_histogram_adds(self): def test_gauge_all(self): g1 = Gauge('g', 'help', registry=None) - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) g2 = Gauge('g', 'help', registry=None) self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) @@ -88,7 +89,7 @@ def test_gauge_all(self): def test_gauge_liveall(self): g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) g2 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) @@ -102,7 +103,7 @@ def test_gauge_liveall(self): def test_gauge_min(self): g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min') - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min') self.assertEqual(0, self.registry.get_sample_value('g')) g1.set(1) @@ -111,7 +112,7 @@ def test_gauge_min(self): def test_gauge_max(self): g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max') - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max') self.assertEqual(0, self.registry.get_sample_value('g')) g1.set(1) @@ -120,7 +121,7 @@ def test_gauge_max(self): def test_gauge_livesum(self): g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') - values.ValueClass = MultiProcessValue(lambda: 456) + values.ValueClass = MmapedValue(lambda: 456) g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') self.assertEqual(0, self.registry.get_sample_value('g')) g1.set(1) @@ -136,7 +137,7 @@ def test_namespace_subsystem(self): def test_counter_across_forks(self): pid = 0 - values.ValueClass = MultiProcessValue(lambda: pid) + values.ValueClass = MmapedValue(lambda: pid) c1 = Counter('c', 'help', registry=None) self.assertEqual(0, self.registry.get_sample_value('c_total')) c1.inc(1) @@ -148,7 +149,7 @@ def test_counter_across_forks(self): def test_initialization_detects_pid_change(self): pid = 0 - values.ValueClass = MultiProcessValue(lambda: pid) + values.ValueClass = MmapedValue(lambda: pid) # can not inspect the files cache directly, as it's a closure, so we # check for the actual files themselves @@ -169,7 +170,7 @@ def files(): @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") def test_collect(self): pid = 0 - values.ValueClass = MultiProcessValue(lambda: pid) + values.ValueClass = MmapedValue(lambda: pid) labels = dict((i, i) for i in 'abcd') def add_label(key, value): @@ -230,7 +231,7 @@ def add_label(key, value): @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") def test_merge_no_accumulate(self): pid = 0 - values.ValueClass = MultiProcessValue(lambda: pid) + values.ValueClass = MmapedValue(lambda: pid) labels = dict((i, i) for i in 'abcd') def add_label(key, value): From c02c3758f4f4ea2db634a5cc64432cf470897386 Mon Sep 17 00:00:00 2001 From: Aarni Koskela Date: Mon, 21 Jan 2019 15:21:09 +0200 Subject: [PATCH 2/6] Prepare for multiple multiprocess backends Signed-off-by: Aarni Koskela --- prometheus_client/multiprocess/__init__.py | 134 ++------ .../multiprocess/mmap_collector.py | 109 ++++++ prometheus_client/multiprocess/mmap_utils.py | 9 + .../multiprocess/mmaped_value.py | 3 +- prometheus_client/values.py | 6 +- tests/test_multiprocess.py | 312 +---------------- tests/test_multiprocess_mmap.py | 313 ++++++++++++++++++ 7 files changed, 461 insertions(+), 425 deletions(-) create mode 100644 prometheus_client/multiprocess/mmap_collector.py create mode 100644 prometheus_client/multiprocess/mmap_utils.py create mode 100644 tests/test_multiprocess_mmap.py diff --git a/prometheus_client/multiprocess/__init__.py b/prometheus_client/multiprocess/__init__.py index 449de748..8388f97d 100644 --- a/prometheus_client/multiprocess/__init__.py +++ b/prometheus_client/multiprocess/__init__.py @@ -1,125 +1,37 @@ from __future__ import unicode_literals -from collections import defaultdict -import glob -import json import os -from ..metrics_core import Metric -from .mmap_dict import MmapedDict -from ..samples import Sample -from ..utils import floatToGoString +multiprocess_enabled = bool(os.environ.get('prometheus_multiproc_dir')) +multiprocess_backend = (os.environ.get('prometheus_multiproc_backend') or 'mmap') -class MultiProcessCollector(object): - """Collector for files for multi-process mode.""" - - def __init__(self, registry, path=None): - if path is None: - path = os.environ.get('prometheus_multiproc_dir') - if not path or not os.path.isdir(path): - raise ValueError('env prometheus_multiproc_dir is not set or not a directory') - self._path = path - if registry: - registry.register(self) - - def collect(self): - files = glob.glob(os.path.join(self._path, '*.db')) - return self.merge(files, accumulate=True) - - def merge(self, files, accumulate=True): - """Merge metrics from given mmap files. - - By default, histograms are accumulated, as per prometheus wire format. - But if writing the merged data back to mmap files, use - accumulate=False to avoid compound accumulation. - """ - metrics = {} - for f in files: - parts = os.path.basename(f).split('_') - typ = parts[0] - d = MmapedDict(f, read_mode=True) - for key, value in d.read_all_values(): - metric_name, name, labels = json.loads(key) - labels_key = tuple(sorted(labels.items())) - - metric = metrics.get(metric_name) - if metric is None: - metric = Metric(metric_name, 'Multiprocess metric', typ) - metrics[metric_name] = metric - - if typ == 'gauge': - pid = parts[2][:-3] - metric._multiprocess_mode = parts[1] - metric.add_sample(name, labels_key + (('pid', pid), ), value) - else: - # The duplicates and labels are fixed in the next for. - metric.add_sample(name, labels_key, value) - d.close() - - for metric in metrics.values(): - samples = defaultdict(float) - buckets = {} - for s in metric.samples: - name, labels, value = s.name, s.labels, s.value - if metric.type == 'gauge': - without_pid = tuple(l for l in labels if l[0] != 'pid') - if metric._multiprocess_mode == 'min': - current = samples.setdefault((name, without_pid), value) - if value < current: - samples[(s.name, without_pid)] = value - elif metric._multiprocess_mode == 'max': - current = samples.setdefault((name, without_pid), value) - if value > current: - samples[(s.name, without_pid)] = value - elif metric._multiprocess_mode == 'livesum': - samples[(name, without_pid)] += value - else: # all/liveall - samples[(name, labels)] = value - - elif metric.type == 'histogram': - bucket = tuple(float(l[1]) for l in labels if l[0] == 'le') - if bucket: - # _bucket - without_le = tuple(l for l in labels if l[0] != 'le') - buckets.setdefault(without_le, {}) - buckets[without_le].setdefault(bucket[0], 0.0) - buckets[without_le][bucket[0]] += value - else: - # _sum/_count - samples[(s.name, labels)] += value - - else: - # Counter and Summary. - samples[(s.name, labels)] += value +def MultiProcessCollector(registry, path=None): + if path is None: + path = os.environ.get('prometheus_multiproc_dir') + if not path or not os.path.isdir(path): + raise ValueError('env prometheus_multiproc_dir is not set or %s is not a directory' % path) + if multiprocess_backend == 'mmap': + from prometheus_client.multiprocess.mmap_collector import MmapMultiProcessCollector + return MmapMultiProcessCollector(registry=registry, path=path) + else: + raise NotImplementedError('unknown multiprocess backend %s' % multiprocess_backend) - # Accumulate bucket values. - if metric.type == 'histogram': - for labels, values in buckets.items(): - acc = 0.0 - for bucket, value in sorted(values.items()): - sample_key = ( - metric.name + '_bucket', - labels + (('le', floatToGoString(bucket)), ), - ) - if accumulate: - acc += value - samples[sample_key] = acc - else: - samples[sample_key] = value - if accumulate: - samples[(metric.name + '_count', labels)] = acc - # Convert to correct sample format. - metric.samples = [Sample(name, dict(labels), value) for (name, labels), value in samples.items()] - return metrics.values() +def get_multiprocess_value_class(): + if multiprocess_backend == 'mmap': + from prometheus_client.multiprocess.mmaped_value import MmapedValue + return MmapedValue() + else: + raise NotImplementedError('unknown multiprocess backend %s' % multiprocess_backend) def mark_process_dead(pid, path=None): """Do bookkeeping for when one process dies in a multi-process setup.""" if path is None: path = os.environ.get('prometheus_multiproc_dir') - for f in glob.glob(os.path.join(path, 'gauge_livesum_{0}.db'.format(pid))): - os.remove(f) - for f in glob.glob(os.path.join(path, 'gauge_liveall_{0}.db'.format(pid))): - os.remove(f) + if multiprocess_backend == 'mmap': + from prometheus_client.multiprocess.mmap_utils import mmap_cleanup + mmap_cleanup(path, pid) + else: + raise NotImplementedError('unknown multiprocess backend %s' % multiprocess_backend) diff --git a/prometheus_client/multiprocess/mmap_collector.py b/prometheus_client/multiprocess/mmap_collector.py new file mode 100644 index 00000000..d86d457c --- /dev/null +++ b/prometheus_client/multiprocess/mmap_collector.py @@ -0,0 +1,109 @@ +import glob +import json +import os +from collections import defaultdict + +from prometheus_client.metrics_core import Metric +from prometheus_client.multiprocess.mmap_dict import MmapedDict +from prometheus_client.samples import Sample +from prometheus_client.utils import floatToGoString + + +class MmapMultiProcessCollector(object): + """Collector for files for multi-process mode.""" + + def __init__(self, registry, path): + self._path = path + if registry: + registry.register(self) + + def collect(self): + files = glob.glob(os.path.join(self._path, '*.db')) + return self.merge(files, accumulate=True) + + def merge(self, files, accumulate=True): + """Merge metrics from given mmap files. + + By default, histograms are accumulated, as per prometheus wire format. + But if writing the merged data back to mmap files, use + accumulate=False to avoid compound accumulation. + """ + metrics = {} + for f in files: + parts = os.path.basename(f).split('_') + typ = parts[0] + d = MmapedDict(f, read_mode=True) + for key, value in d.read_all_values(): + metric_name, name, labels = json.loads(key) + labels_key = tuple(sorted(labels.items())) + + metric = metrics.get(metric_name) + if metric is None: + metric = Metric(metric_name, 'Multiprocess metric', typ) + metrics[metric_name] = metric + + if typ == 'gauge': + pid = parts[2][:-3] + metric._multiprocess_mode = parts[1] + metric.add_sample(name, labels_key + (('pid', pid), ), value) + else: + # The duplicates and labels are fixed in the next for. + metric.add_sample(name, labels_key, value) + d.close() + + for metric in metrics.values(): + samples = defaultdict(float) + buckets = {} + for s in metric.samples: + name, labels, value = s.name, s.labels, s.value + if metric.type == 'gauge': + without_pid = tuple(l for l in labels if l[0] != 'pid') + if metric._multiprocess_mode == 'min': + current = samples.setdefault((name, without_pid), value) + if value < current: + samples[(s.name, without_pid)] = value + elif metric._multiprocess_mode == 'max': + current = samples.setdefault((name, without_pid), value) + if value > current: + samples[(s.name, without_pid)] = value + elif metric._multiprocess_mode == 'livesum': + samples[(name, without_pid)] += value + else: # all/liveall + samples[(name, labels)] = value + + elif metric.type == 'histogram': + bucket = tuple(float(l[1]) for l in labels if l[0] == 'le') + if bucket: + # _bucket + without_le = tuple(l for l in labels if l[0] != 'le') + buckets.setdefault(without_le, {}) + buckets[without_le].setdefault(bucket[0], 0.0) + buckets[without_le][bucket[0]] += value + else: + # _sum/_count + samples[(s.name, labels)] += value + + else: + # Counter and Summary. + samples[(s.name, labels)] += value + + # Accumulate bucket values. + if metric.type == 'histogram': + for labels, values in buckets.items(): + acc = 0.0 + for bucket, value in sorted(values.items()): + sample_key = ( + metric.name + '_bucket', + labels + (('le', floatToGoString(bucket)), ), + ) + if accumulate: + acc += value + samples[sample_key] = acc + else: + samples[sample_key] = value + if accumulate: + samples[(metric.name + '_count', labels)] = acc + + # Convert to correct sample format. + metric.samples = [Sample(name, dict(labels), value) for (name, labels), value in samples.items()] + return metrics.values() diff --git a/prometheus_client/multiprocess/mmap_utils.py b/prometheus_client/multiprocess/mmap_utils.py new file mode 100644 index 00000000..83773d6b --- /dev/null +++ b/prometheus_client/multiprocess/mmap_utils.py @@ -0,0 +1,9 @@ +import glob +import os + + +def mmap_cleanup(path, pid): + for f in glob.glob(os.path.join(path, 'gauge_livesum_{0}.db'.format(pid))): + os.remove(f) + for f in glob.glob(os.path.join(path, 'gauge_liveall_{0}.db'.format(pid))): + os.remove(f) diff --git a/prometheus_client/multiprocess/mmaped_value.py b/prometheus_client/multiprocess/mmaped_value.py index e74da31f..8adc6fea 100644 --- a/prometheus_client/multiprocess/mmaped_value.py +++ b/prometheus_client/multiprocess/mmaped_value.py @@ -1,8 +1,7 @@ import os from threading import Lock -from prometheus_client.multiprocess import MmapedDict -from prometheus_client.multiprocess.mmap_dict import mmap_key +from prometheus_client.multiprocess.mmap_dict import MmapedDict, mmap_key def MmapedValue(_pidFunc=os.getpid): diff --git a/prometheus_client/values.py b/prometheus_client/values.py index 7a63c70a..de4a4ff3 100644 --- a/prometheus_client/values.py +++ b/prometheus_client/values.py @@ -31,9 +31,9 @@ def get_value_class(): # This needs to be chosen before the first metric is constructed, # and as that may be in some arbitrary library the user/admin has # no control over we use an environment variable. - if 'prometheus_multiproc_dir' in os.environ: - from prometheus_client.multiprocess.mmaped_value import MmapedValue - return MmapedValue() + from prometheus_client.multiprocess import multiprocess_enabled, get_multiprocess_value_class + if multiprocess_enabled: + return get_multiprocess_value_class() else: return MutexValue diff --git a/tests/test_multiprocess.py b/tests/test_multiprocess.py index d9f0e699..162ef481 100644 --- a/tests/test_multiprocess.py +++ b/tests/test_multiprocess.py @@ -1,316 +1,10 @@ -from __future__ import unicode_literals - -import glob import os -import shutil -import sys import tempfile +import unittest -from prometheus_client import core, values -from prometheus_client.core import ( - CollectorRegistry, Counter, Gauge, Histogram, Sample, Summary, -) -from prometheus_client.multiprocess import ( - mark_process_dead, MultiProcessCollector, - mmap_dict) -from prometheus_client.values import MutexValue -from prometheus_client.multiprocess.mmaped_value import MmapedValue - -if sys.version_info < (2, 7): - # We need the skip decorators from unittest2 on Python 2.6. - import unittest2 as unittest -else: - import unittest - - - - -class TestMultiProcess(unittest.TestCase): - def setUp(self): - self.tempdir = tempfile.mkdtemp() - os.environ['prometheus_multiproc_dir'] = self.tempdir - values.ValueClass = MmapedValue(lambda: 123) - self.registry = CollectorRegistry() - self.collector = MultiProcessCollector(self.registry, self.tempdir) - - @property - def _value_class(self): - return - - def tearDown(self): - del os.environ['prometheus_multiproc_dir'] - shutil.rmtree(self.tempdir) - values.ValueClass = MutexValue - - def test_counter_adds(self): - c1 = Counter('c', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - c2 = Counter('c', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('c_total')) - c1.inc(1) - c2.inc(2) - self.assertEqual(3, self.registry.get_sample_value('c_total')) - - def test_summary_adds(self): - s1 = Summary('s', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - s2 = Summary('s', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('s_count')) - self.assertEqual(0, self.registry.get_sample_value('s_sum')) - s1.observe(1) - s2.observe(2) - self.assertEqual(2, self.registry.get_sample_value('s_count')) - self.assertEqual(3, self.registry.get_sample_value('s_sum')) - - def test_histogram_adds(self): - h1 = Histogram('h', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - h2 = Histogram('h', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('h_count')) - self.assertEqual(0, self.registry.get_sample_value('h_sum')) - self.assertEqual(0, self.registry.get_sample_value('h_bucket', {'le': '5.0'})) - h1.observe(1) - h2.observe(2) - self.assertEqual(2, self.registry.get_sample_value('h_count')) - self.assertEqual(3, self.registry.get_sample_value('h_sum')) - self.assertEqual(2, self.registry.get_sample_value('h_bucket', {'le': '5.0'})) - - def test_gauge_all(self): - g1 = Gauge('g', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) - g1.set(1) - g2.set(2) - mark_process_dead(123, os.environ['prometheus_multiproc_dir']) - self.assertEqual(1, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) - - def test_gauge_liveall(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) - g1.set(1) - g2.set(2) - self.assertEqual(1, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) - mark_process_dead(123, os.environ['prometheus_multiproc_dir']) - self.assertEqual(None, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) - - def test_gauge_min(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min') - self.assertEqual(0, self.registry.get_sample_value('g')) - g1.set(1) - g2.set(2) - self.assertEqual(1, self.registry.get_sample_value('g')) - - def test_gauge_max(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max') - self.assertEqual(0, self.registry.get_sample_value('g')) - g1.set(1) - g2.set(2) - self.assertEqual(2, self.registry.get_sample_value('g')) - - def test_gauge_livesum(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') - self.assertEqual(0, self.registry.get_sample_value('g')) - g1.set(1) - g2.set(2) - self.assertEqual(3, self.registry.get_sample_value('g')) - mark_process_dead(123, os.environ['prometheus_multiproc_dir']) - self.assertEqual(2, self.registry.get_sample_value('g')) - - def test_namespace_subsystem(self): - c1 = Counter('c', 'help', registry=None, namespace='ns', subsystem='ss') - c1.inc(1) - self.assertEqual(1, self.registry.get_sample_value('ns_ss_c_total')) - - def test_counter_across_forks(self): - pid = 0 - values.ValueClass = MmapedValue(lambda: pid) - c1 = Counter('c', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('c_total')) - c1.inc(1) - c1.inc(1) - pid = 1 - c1.inc(1) - self.assertEqual(3, self.registry.get_sample_value('c_total')) - self.assertEqual(1, c1._value.get()) - - def test_initialization_detects_pid_change(self): - pid = 0 - values.ValueClass = MmapedValue(lambda: pid) - - # can not inspect the files cache directly, as it's a closure, so we - # check for the actual files themselves - def files(): - fs = os.listdir(os.environ['prometheus_multiproc_dir']) - fs.sort() - return fs - - c1 = Counter('c1', 'c1', registry=None) - self.assertEqual(files(), ['counter_0.db']) - c2 = Counter('c2', 'c2', registry=None) - self.assertEqual(files(), ['counter_0.db']) - pid = 1 - c3 = Counter('c3', 'c3', registry=None) - self.assertEqual(files(), ['counter_0.db', 'counter_1.db']) - +from prometheus_client.multiprocess import MultiProcessCollector - @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") - def test_collect(self): - pid = 0 - values.ValueClass = MmapedValue(lambda: pid) - labels = dict((i, i) for i in 'abcd') - - def add_label(key, value): - l = labels.copy() - l[key] = value - return l - - c = Counter('c', 'help', labelnames=labels.keys(), registry=None) - g = Gauge('g', 'help', labelnames=labels.keys(), registry=None) - h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) - - c.labels(**labels).inc(1) - g.labels(**labels).set(1) - h.labels(**labels).observe(1) - - pid = 1 - - c.labels(**labels).inc(1) - g.labels(**labels).set(1) - h.labels(**labels).observe(5) - - metrics = dict((m.name, m) for m in self.collector.collect()) - - self.assertEqual( - metrics['c'].samples, [Sample('c_total', labels, 2.0)] - ) - metrics['g'].samples.sort(key=lambda x: x[1]['pid']) - self.assertEqual(metrics['g'].samples, [ - Sample('g', add_label('pid', '0'), 1.0), - Sample('g', add_label('pid', '1'), 1.0), - ]) - - metrics['h'].samples.sort( - key=lambda x: (x[0], float(x[1].get('le', 0))) - ) - expected_histogram = [ - Sample('h_bucket', add_label('le', '0.005'), 0.0), - Sample('h_bucket', add_label('le', '0.01'), 0.0), - Sample('h_bucket', add_label('le', '0.025'), 0.0), - Sample('h_bucket', add_label('le', '0.05'), 0.0), - Sample('h_bucket', add_label('le', '0.075'), 0.0), - Sample('h_bucket', add_label('le', '0.1'), 0.0), - Sample('h_bucket', add_label('le', '0.25'), 0.0), - Sample('h_bucket', add_label('le', '0.5'), 0.0), - Sample('h_bucket', add_label('le', '0.75'), 0.0), - Sample('h_bucket', add_label('le', '1.0'), 1.0), - Sample('h_bucket', add_label('le', '2.5'), 1.0), - Sample('h_bucket', add_label('le', '5.0'), 2.0), - Sample('h_bucket', add_label('le', '7.5'), 2.0), - Sample('h_bucket', add_label('le', '10.0'), 2.0), - Sample('h_bucket', add_label('le', '+Inf'), 2.0), - Sample('h_count', labels, 2.0), - Sample('h_sum', labels, 6.0), - ] - - self.assertEqual(metrics['h'].samples, expected_histogram) - - @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") - def test_merge_no_accumulate(self): - pid = 0 - values.ValueClass = MmapedValue(lambda: pid) - labels = dict((i, i) for i in 'abcd') - - def add_label(key, value): - l = labels.copy() - l[key] = value - return l - - h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) - h.labels(**labels).observe(1) - pid = 1 - h.labels(**labels).observe(5) - - path = os.path.join(os.environ['prometheus_multiproc_dir'], '*.db') - files = glob.glob(path) - metrics = dict( - (m.name, m) for m in self.collector.merge(files, accumulate=False) - ) - - metrics['h'].samples.sort( - key=lambda x: (x[0], float(x[1].get('le', 0))) - ) - expected_histogram = [ - Sample('h_bucket', add_label('le', '0.005'), 0.0), - Sample('h_bucket', add_label('le', '0.01'), 0.0), - Sample('h_bucket', add_label('le', '0.025'), 0.0), - Sample('h_bucket', add_label('le', '0.05'), 0.0), - Sample('h_bucket', add_label('le', '0.075'), 0.0), - Sample('h_bucket', add_label('le', '0.1'), 0.0), - Sample('h_bucket', add_label('le', '0.25'), 0.0), - Sample('h_bucket', add_label('le', '0.5'), 0.0), - Sample('h_bucket', add_label('le', '0.75'), 0.0), - Sample('h_bucket', add_label('le', '1.0'), 1.0), - Sample('h_bucket', add_label('le', '2.5'), 0.0), - Sample('h_bucket', add_label('le', '5.0'), 1.0), - Sample('h_bucket', add_label('le', '7.5'), 0.0), - Sample('h_bucket', add_label('le', '10.0'), 0.0), - Sample('h_bucket', add_label('le', '+Inf'), 0.0), - Sample('h_sum', labels, 6.0), - ] - - self.assertEqual(metrics['h'].samples, expected_histogram) - - -class TestMmapedDict(unittest.TestCase): - def setUp(self): - fd, self.tempfile = tempfile.mkstemp() - os.close(fd) - self.d = mmap_dict.MmapedDict(self.tempfile) - - def test_process_restart(self): - self.d.write_value('abc', 123.0) - self.d.close() - self.d = mmap_dict.MmapedDict(self.tempfile) - self.assertEqual(123, self.d.read_value('abc')) - self.assertEqual([('abc', 123.0)], list(self.d.read_all_values())) - - def test_expansion(self): - key = 'a' * mmap_dict._INITIAL_MMAP_SIZE - self.d.write_value(key, 123.0) - self.assertEqual([(key, 123.0)], list(self.d.read_all_values())) - - def test_multi_expansion(self): - key = 'a' * mmap_dict._INITIAL_MMAP_SIZE * 4 - self.d.write_value('abc', 42.0) - self.d.write_value(key, 123.0) - self.d.write_value('def', 17.0) - self.assertEqual( - [('abc', 42.0), (key, 123.0), ('def', 17.0)], - list(self.d.read_all_values())) - - def test_corruption_detected(self): - self.d.write_value('abc', 42.0) - # corrupt the written data - self.d._m[8:16] = b'somejunk' - with self.assertRaises(RuntimeError): - list(self.d.read_all_values()) - - def tearDown(self): - os.unlink(self.tempfile) +from prometheus_client.registry import CollectorRegistry class TestUnsetEnv(unittest.TestCase): diff --git a/tests/test_multiprocess_mmap.py b/tests/test_multiprocess_mmap.py new file mode 100644 index 00000000..c15354cc --- /dev/null +++ b/tests/test_multiprocess_mmap.py @@ -0,0 +1,313 @@ +from __future__ import unicode_literals + +import glob +import os +import shutil +import sys +import tempfile + +from prometheus_client import values +from prometheus_client.core import ( + CollectorRegistry, Counter, Gauge, Histogram, Sample, Summary, +) +from prometheus_client.multiprocess import mark_process_dead, mmap_dict, mmap_collector +from prometheus_client.values import MutexValue +from prometheus_client.multiprocess.mmaped_value import MmapedValue + +if sys.version_info < (2, 7): + # We need the skip decorators from unittest2 on Python 2.6. + import unittest2 as unittest +else: + import unittest + + + + +class TestMmapMultiProcess(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.mkdtemp() + os.environ['prometheus_multiproc_dir'] = self.tempdir + values.ValueClass = MmapedValue(lambda: 123) + self.registry = CollectorRegistry() + self.collector = mmap_collector.MmapMultiProcessCollector(self.registry, path=self.tempdir) + + @property + def _value_class(self): + return + + def tearDown(self): + del os.environ['prometheus_multiproc_dir'] + shutil.rmtree(self.tempdir) + values.ValueClass = MutexValue + + def test_counter_adds(self): + c1 = Counter('c', 'help', registry=None) + values.ValueClass = MmapedValue(lambda: 456) + c2 = Counter('c', 'help', registry=None) + self.assertEqual(0, self.registry.get_sample_value('c_total')) + c1.inc(1) + c2.inc(2) + self.assertEqual(3, self.registry.get_sample_value('c_total')) + + def test_summary_adds(self): + s1 = Summary('s', 'help', registry=None) + values.ValueClass = MmapedValue(lambda: 456) + s2 = Summary('s', 'help', registry=None) + self.assertEqual(0, self.registry.get_sample_value('s_count')) + self.assertEqual(0, self.registry.get_sample_value('s_sum')) + s1.observe(1) + s2.observe(2) + self.assertEqual(2, self.registry.get_sample_value('s_count')) + self.assertEqual(3, self.registry.get_sample_value('s_sum')) + + def test_histogram_adds(self): + h1 = Histogram('h', 'help', registry=None) + values.ValueClass = MmapedValue(lambda: 456) + h2 = Histogram('h', 'help', registry=None) + self.assertEqual(0, self.registry.get_sample_value('h_count')) + self.assertEqual(0, self.registry.get_sample_value('h_sum')) + self.assertEqual(0, self.registry.get_sample_value('h_bucket', {'le': '5.0'})) + h1.observe(1) + h2.observe(2) + self.assertEqual(2, self.registry.get_sample_value('h_count')) + self.assertEqual(3, self.registry.get_sample_value('h_sum')) + self.assertEqual(2, self.registry.get_sample_value('h_bucket', {'le': '5.0'})) + + def test_gauge_all(self): + g1 = Gauge('g', 'help', registry=None) + values.ValueClass = MmapedValue(lambda: 456) + g2 = Gauge('g', 'help', registry=None) + self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) + self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) + g1.set(1) + g2.set(2) + mark_process_dead(123, os.environ['prometheus_multiproc_dir']) + self.assertEqual(1, self.registry.get_sample_value('g', {'pid': '123'})) + self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) + + def test_gauge_liveall(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') + values.ValueClass = MmapedValue(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') + self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) + self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) + g1.set(1) + g2.set(2) + self.assertEqual(1, self.registry.get_sample_value('g', {'pid': '123'})) + self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) + mark_process_dead(123, os.environ['prometheus_multiproc_dir']) + self.assertEqual(None, self.registry.get_sample_value('g', {'pid': '123'})) + self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) + + def test_gauge_min(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min') + values.ValueClass = MmapedValue(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min') + self.assertEqual(0, self.registry.get_sample_value('g')) + g1.set(1) + g2.set(2) + self.assertEqual(1, self.registry.get_sample_value('g')) + + def test_gauge_max(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max') + values.ValueClass = MmapedValue(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max') + self.assertEqual(0, self.registry.get_sample_value('g')) + g1.set(1) + g2.set(2) + self.assertEqual(2, self.registry.get_sample_value('g')) + + def test_gauge_livesum(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') + values.ValueClass = MmapedValue(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') + self.assertEqual(0, self.registry.get_sample_value('g')) + g1.set(1) + g2.set(2) + self.assertEqual(3, self.registry.get_sample_value('g')) + mark_process_dead(123, os.environ['prometheus_multiproc_dir']) + self.assertEqual(2, self.registry.get_sample_value('g')) + + def test_namespace_subsystem(self): + c1 = Counter('c', 'help', registry=None, namespace='ns', subsystem='ss') + c1.inc(1) + self.assertEqual(1, self.registry.get_sample_value('ns_ss_c_total')) + + def test_counter_across_forks(self): + pid = 0 + values.ValueClass = MmapedValue(lambda: pid) + c1 = Counter('c', 'help', registry=None) + self.assertEqual(0, self.registry.get_sample_value('c_total')) + c1.inc(1) + c1.inc(1) + pid = 1 + c1.inc(1) + self.assertEqual(3, self.registry.get_sample_value('c_total')) + self.assertEqual(1, c1._value.get()) + + def test_initialization_detects_pid_change(self): + pid = 0 + values.ValueClass = MmapedValue(lambda: pid) + + # can not inspect the files cache directly, as it's a closure, so we + # check for the actual files themselves + def files(): + fs = os.listdir(os.environ['prometheus_multiproc_dir']) + fs.sort() + return fs + + c1 = Counter('c1', 'c1', registry=None) + self.assertEqual(files(), ['counter_0.db']) + c2 = Counter('c2', 'c2', registry=None) + self.assertEqual(files(), ['counter_0.db']) + pid = 1 + c3 = Counter('c3', 'c3', registry=None) + self.assertEqual(files(), ['counter_0.db', 'counter_1.db']) + + + @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") + def test_collect(self): + pid = 0 + values.ValueClass = MmapedValue(lambda: pid) + labels = dict((i, i) for i in 'abcd') + + def add_label(key, value): + l = labels.copy() + l[key] = value + return l + + c = Counter('c', 'help', labelnames=labels.keys(), registry=None) + g = Gauge('g', 'help', labelnames=labels.keys(), registry=None) + h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) + + c.labels(**labels).inc(1) + g.labels(**labels).set(1) + h.labels(**labels).observe(1) + + pid = 1 + + c.labels(**labels).inc(1) + g.labels(**labels).set(1) + h.labels(**labels).observe(5) + + metrics = dict((m.name, m) for m in self.collector.collect()) + + self.assertEqual( + metrics['c'].samples, [Sample('c_total', labels, 2.0)] + ) + metrics['g'].samples.sort(key=lambda x: x[1]['pid']) + self.assertEqual(metrics['g'].samples, [ + Sample('g', add_label('pid', '0'), 1.0), + Sample('g', add_label('pid', '1'), 1.0), + ]) + + metrics['h'].samples.sort( + key=lambda x: (x[0], float(x[1].get('le', 0))) + ) + expected_histogram = [ + Sample('h_bucket', add_label('le', '0.005'), 0.0), + Sample('h_bucket', add_label('le', '0.01'), 0.0), + Sample('h_bucket', add_label('le', '0.025'), 0.0), + Sample('h_bucket', add_label('le', '0.05'), 0.0), + Sample('h_bucket', add_label('le', '0.075'), 0.0), + Sample('h_bucket', add_label('le', '0.1'), 0.0), + Sample('h_bucket', add_label('le', '0.25'), 0.0), + Sample('h_bucket', add_label('le', '0.5'), 0.0), + Sample('h_bucket', add_label('le', '0.75'), 0.0), + Sample('h_bucket', add_label('le', '1.0'), 1.0), + Sample('h_bucket', add_label('le', '2.5'), 1.0), + Sample('h_bucket', add_label('le', '5.0'), 2.0), + Sample('h_bucket', add_label('le', '7.5'), 2.0), + Sample('h_bucket', add_label('le', '10.0'), 2.0), + Sample('h_bucket', add_label('le', '+Inf'), 2.0), + Sample('h_count', labels, 2.0), + Sample('h_sum', labels, 6.0), + ] + + self.assertEqual(metrics['h'].samples, expected_histogram) + + @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") + def test_merge_no_accumulate(self): + pid = 0 + values.ValueClass = MmapedValue(lambda: pid) + labels = dict((i, i) for i in 'abcd') + + def add_label(key, value): + l = labels.copy() + l[key] = value + return l + + h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) + h.labels(**labels).observe(1) + pid = 1 + h.labels(**labels).observe(5) + + path = os.path.join(os.environ['prometheus_multiproc_dir'], '*.db') + files = glob.glob(path) + metrics = dict( + (m.name, m) for m in self.collector.merge(files, accumulate=False) + ) + + metrics['h'].samples.sort( + key=lambda x: (x[0], float(x[1].get('le', 0))) + ) + expected_histogram = [ + Sample('h_bucket', add_label('le', '0.005'), 0.0), + Sample('h_bucket', add_label('le', '0.01'), 0.0), + Sample('h_bucket', add_label('le', '0.025'), 0.0), + Sample('h_bucket', add_label('le', '0.05'), 0.0), + Sample('h_bucket', add_label('le', '0.075'), 0.0), + Sample('h_bucket', add_label('le', '0.1'), 0.0), + Sample('h_bucket', add_label('le', '0.25'), 0.0), + Sample('h_bucket', add_label('le', '0.5'), 0.0), + Sample('h_bucket', add_label('le', '0.75'), 0.0), + Sample('h_bucket', add_label('le', '1.0'), 1.0), + Sample('h_bucket', add_label('le', '2.5'), 0.0), + Sample('h_bucket', add_label('le', '5.0'), 1.0), + Sample('h_bucket', add_label('le', '7.5'), 0.0), + Sample('h_bucket', add_label('le', '10.0'), 0.0), + Sample('h_bucket', add_label('le', '+Inf'), 0.0), + Sample('h_sum', labels, 6.0), + ] + + self.assertEqual(metrics['h'].samples, expected_histogram) + + +class TestMmapedDict(unittest.TestCase): + def setUp(self): + fd, self.tempfile = tempfile.mkstemp() + os.close(fd) + self.d = mmap_dict.MmapedDict(self.tempfile) + + def test_process_restart(self): + self.d.write_value('abc', 123.0) + self.d.close() + self.d = mmap_dict.MmapedDict(self.tempfile) + self.assertEqual(123, self.d.read_value('abc')) + self.assertEqual([('abc', 123.0)], list(self.d.read_all_values())) + + def test_expansion(self): + key = 'a' * mmap_dict._INITIAL_MMAP_SIZE + self.d.write_value(key, 123.0) + self.assertEqual([(key, 123.0)], list(self.d.read_all_values())) + + def test_multi_expansion(self): + key = 'a' * mmap_dict._INITIAL_MMAP_SIZE * 4 + self.d.write_value('abc', 42.0) + self.d.write_value(key, 123.0) + self.d.write_value('def', 17.0) + self.assertEqual( + [('abc', 42.0), (key, 123.0), ('def', 17.0)], + list(self.d.read_all_values())) + + def test_corruption_detected(self): + self.d.write_value('abc', 42.0) + # corrupt the written data + self.d._m[8:16] = b'somejunk' + with self.assertRaises(RuntimeError): + list(self.d.read_all_values()) + + def tearDown(self): + os.unlink(self.tempfile) + + From 98f6d79081a723243d9b1569d1317815a720dbfb Mon Sep 17 00:00:00 2001 From: Aarni Koskela Date: Mon, 21 Jan 2019 16:28:33 +0200 Subject: [PATCH 3/6] Hoist most of multiprocess tests up to common base class Signed-off-by: Aarni Koskela --- tests/common_multiprocess_tests.py | 199 ++++++++++++++++++++++++++ tests/test_multiprocess_mmap.py | 222 +++-------------------------- 2 files changed, 216 insertions(+), 205 deletions(-) create mode 100644 tests/common_multiprocess_tests.py diff --git a/tests/common_multiprocess_tests.py b/tests/common_multiprocess_tests.py new file mode 100644 index 00000000..e2041d44 --- /dev/null +++ b/tests/common_multiprocess_tests.py @@ -0,0 +1,199 @@ +from __future__ import unicode_literals + +import os +import shutil +import sys +import tempfile + +import pytest + +from prometheus_client import values +from prometheus_client.core import ( + CollectorRegistry, Counter, Gauge, Histogram, Sample, Summary, +) +import prometheus_client.multiprocess as multiprocess + + +class CommonMultiprocessTests: + + _multiprocess_backend = None + _collector_class = None + _value_class = None + + def setUp(self): + multiprocess.multiprocess_backend = self._multiprocess_backend + self.tempdir = tempfile.mkdtemp() + os.environ['prometheus_multiproc_dir'] = self.tempdir + values.ValueClass = self._value_class(lambda: 123) + self.registry = CollectorRegistry() + self.collector = self._collector_class(self.registry, path=self.tempdir) + + def tearDown(self): + del os.environ['prometheus_multiproc_dir'] + shutil.rmtree(self.tempdir) + multiprocess.multiprocess_backend = None + values.ValueClass = values.MutexValue + + def test_counter_adds(self): + c1 = Counter('c', 'help', registry=None) + values.ValueClass = self._value_class(lambda: 456) + c2 = Counter('c', 'help', registry=None) + assert self.registry.get_sample_value('c_total') == 0 + c1.inc(1) + c2.inc(2) + assert self.registry.get_sample_value('c_total') == 3 + + def test_summary_adds(self): + s1 = Summary('s', 'help', registry=None) + values.ValueClass = self._value_class(lambda: 456) + s2 = Summary('s', 'help', registry=None) + assert self.registry.get_sample_value('s_count') == 0 + assert self.registry.get_sample_value('s_sum') == 0 + s1.observe(1) + s2.observe(2) + assert self.registry.get_sample_value('s_count') == 2 + assert self.registry.get_sample_value('s_sum') == 3 + + def test_histogram_adds(self): + h1 = Histogram('h', 'help', registry=None) + values.ValueClass = self._value_class(lambda: 456) + h2 = Histogram('h', 'help', registry=None) + assert self.registry.get_sample_value('h_count') == 0 + assert self.registry.get_sample_value('h_sum') == 0 + assert self.registry.get_sample_value('h_bucket', {'le': '5.0'}) == 0 + h1.observe(1) + h2.observe(2) + assert self.registry.get_sample_value('h_count') == 2 + assert self.registry.get_sample_value('h_sum') == 3 + assert self.registry.get_sample_value('h_bucket', {'le': '5.0'}) == 2 + + def test_gauge_all(self): + g1 = Gauge('g', 'help', registry=None) + values.ValueClass = self._value_class(lambda: 456) + g2 = Gauge('g', 'help', registry=None) + assert self.registry.get_sample_value('g', {'pid': '123'}) == 0 + assert self.registry.get_sample_value('g', {'pid': '456'}) == 0 + g1.set(1) + g2.set(2) + multiprocess.mark_process_dead(123, os.environ['prometheus_multiproc_dir']) + assert self.registry.get_sample_value('g', {'pid': '123'}) == 1 + assert self.registry.get_sample_value('g', {'pid': '456'}) == 2 + + def test_gauge_liveall(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') + values.ValueClass = self._value_class(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') + assert self.registry.get_sample_value('g', {'pid': '123'}) == 0 + assert self.registry.get_sample_value('g', {'pid': '456'}) == 0 + g1.set(1) + g2.set(2) + assert self.registry.get_sample_value('g', {'pid': '123'}) == 1 + assert self.registry.get_sample_value('g', {'pid': '456'}) == 2 + multiprocess.mark_process_dead(123, os.environ['prometheus_multiproc_dir']) + assert self.registry.get_sample_value('g', {'pid': '123'}) == None + assert self.registry.get_sample_value('g', {'pid': '456'}) == 2 + + def test_gauge_min(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min') + values.ValueClass = self._value_class(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min') + assert self.registry.get_sample_value('g') == 0 + g1.set(1) + g2.set(2) + assert self.registry.get_sample_value('g') == 1 + + def test_gauge_max(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max') + values.ValueClass = self._value_class(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max') + assert self.registry.get_sample_value('g') == 0 + g1.set(1) + g2.set(2) + assert self.registry.get_sample_value('g') == 2 + + def test_gauge_livesum(self): + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') + values.ValueClass = self._value_class(lambda: 456) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') + assert self.registry.get_sample_value('g') == 0 + g1.set(1) + g2.set(2) + assert self.registry.get_sample_value('g') == 3 + multiprocess.mark_process_dead(123, os.environ['prometheus_multiproc_dir']) + assert self.registry.get_sample_value('g') == 2 + + def test_namespace_subsystem(self): + c1 = Counter('c', 'help', registry=None, namespace='ns', subsystem='ss') + c1.inc(1) + assert self.registry.get_sample_value('ns_ss_c_total') == 1 + + def test_counter_across_forks(self): + pid = 0 + values.ValueClass = self._value_class(lambda: pid) + c1 = Counter('c', 'help', registry=None) + assert self.registry.get_sample_value('c_total') == 0 + c1.inc(1) + c1.inc(1) + pid = 1 + c1.inc(1) + assert self.registry.get_sample_value('c_total') == 3 + assert c1._value.get() == 1 + + @pytest.mark.skipif(sys.version_info < (2, 7), reason="Test requires Python 2.7+.") + def test_collect(self): + pid = 0 + values.ValueClass = self._value_class(lambda: pid) + labels = dict((i, i) for i in 'abcd') + + def add_label(key, value): + l = labels.copy() + l[key] = value + return l + + c = Counter('c', 'help', labelnames=labels.keys(), registry=None) + g = Gauge('g', 'help', labelnames=labels.keys(), registry=None) + h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) + + c.labels(**labels).inc(1) + g.labels(**labels).set(1) + h.labels(**labels).observe(1) + + pid = 1 + + c.labels(**labels).inc(1) + g.labels(**labels).set(1) + h.labels(**labels).observe(5) + + metrics = dict((m.name, m) for m in self.collector.collect()) + + assert metrics['c'].samples == [Sample('c_total', labels, 2.0)] + metrics['g'].samples.sort(key=lambda x: x[1]['pid']) + assert metrics['g'].samples == [ + Sample('g', add_label('pid', '0'), 1.0), + Sample('g', add_label('pid', '1'), 1.0), + ] + + metrics['h'].samples.sort( + key=lambda x: (x[0], float(x[1].get('le', 0))) + ) + expected_histogram = [ + Sample('h_bucket', add_label('le', '0.005'), 0.0), + Sample('h_bucket', add_label('le', '0.01'), 0.0), + Sample('h_bucket', add_label('le', '0.025'), 0.0), + Sample('h_bucket', add_label('le', '0.05'), 0.0), + Sample('h_bucket', add_label('le', '0.075'), 0.0), + Sample('h_bucket', add_label('le', '0.1'), 0.0), + Sample('h_bucket', add_label('le', '0.25'), 0.0), + Sample('h_bucket', add_label('le', '0.5'), 0.0), + Sample('h_bucket', add_label('le', '0.75'), 0.0), + Sample('h_bucket', add_label('le', '1.0'), 1.0), + Sample('h_bucket', add_label('le', '2.5'), 1.0), + Sample('h_bucket', add_label('le', '5.0'), 2.0), + Sample('h_bucket', add_label('le', '7.5'), 2.0), + Sample('h_bucket', add_label('le', '10.0'), 2.0), + Sample('h_bucket', add_label('le', '+Inf'), 2.0), + Sample('h_count', labels, 2.0), + Sample('h_sum', labels, 6.0), + ] + + assert expected_histogram == metrics['h'].samples diff --git a/tests/test_multiprocess_mmap.py b/tests/test_multiprocess_mmap.py index c15354cc..b67e9750 100644 --- a/tests/test_multiprocess_mmap.py +++ b/tests/test_multiprocess_mmap.py @@ -2,152 +2,28 @@ import glob import os -import shutil import sys import tempfile +import unittest -from prometheus_client import values -from prometheus_client.core import ( - CollectorRegistry, Counter, Gauge, Histogram, Sample, Summary, -) -from prometheus_client.multiprocess import mark_process_dead, mmap_dict, mmap_collector -from prometheus_client.values import MutexValue -from prometheus_client.multiprocess.mmaped_value import MmapedValue - -if sys.version_info < (2, 7): - # We need the skip decorators from unittest2 on Python 2.6. - import unittest2 as unittest -else: - import unittest - - - - -class TestMmapMultiProcess(unittest.TestCase): - def setUp(self): - self.tempdir = tempfile.mkdtemp() - os.environ['prometheus_multiproc_dir'] = self.tempdir - values.ValueClass = MmapedValue(lambda: 123) - self.registry = CollectorRegistry() - self.collector = mmap_collector.MmapMultiProcessCollector(self.registry, path=self.tempdir) - - @property - def _value_class(self): - return - - def tearDown(self): - del os.environ['prometheus_multiproc_dir'] - shutil.rmtree(self.tempdir) - values.ValueClass = MutexValue - - def test_counter_adds(self): - c1 = Counter('c', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - c2 = Counter('c', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('c_total')) - c1.inc(1) - c2.inc(2) - self.assertEqual(3, self.registry.get_sample_value('c_total')) - - def test_summary_adds(self): - s1 = Summary('s', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - s2 = Summary('s', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('s_count')) - self.assertEqual(0, self.registry.get_sample_value('s_sum')) - s1.observe(1) - s2.observe(2) - self.assertEqual(2, self.registry.get_sample_value('s_count')) - self.assertEqual(3, self.registry.get_sample_value('s_sum')) - - def test_histogram_adds(self): - h1 = Histogram('h', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - h2 = Histogram('h', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('h_count')) - self.assertEqual(0, self.registry.get_sample_value('h_sum')) - self.assertEqual(0, self.registry.get_sample_value('h_bucket', {'le': '5.0'})) - h1.observe(1) - h2.observe(2) - self.assertEqual(2, self.registry.get_sample_value('h_count')) - self.assertEqual(3, self.registry.get_sample_value('h_sum')) - self.assertEqual(2, self.registry.get_sample_value('h_bucket', {'le': '5.0'})) - - def test_gauge_all(self): - g1 = Gauge('g', 'help', registry=None) - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) - g1.set(1) - g2.set(2) - mark_process_dead(123, os.environ['prometheus_multiproc_dir']) - self.assertEqual(1, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) - - def test_gauge_liveall(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall') - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'})) - g1.set(1) - g2.set(2) - self.assertEqual(1, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) - mark_process_dead(123, os.environ['prometheus_multiproc_dir']) - self.assertEqual(None, self.registry.get_sample_value('g', {'pid': '123'})) - self.assertEqual(2, self.registry.get_sample_value('g', {'pid': '456'})) - - def test_gauge_min(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min') - self.assertEqual(0, self.registry.get_sample_value('g')) - g1.set(1) - g2.set(2) - self.assertEqual(1, self.registry.get_sample_value('g')) +import pytest - def test_gauge_max(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max') - self.assertEqual(0, self.registry.get_sample_value('g')) - g1.set(1) - g2.set(2) - self.assertEqual(2, self.registry.get_sample_value('g')) - - def test_gauge_livesum(self): - g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') - values.ValueClass = MmapedValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum') - self.assertEqual(0, self.registry.get_sample_value('g')) - g1.set(1) - g2.set(2) - self.assertEqual(3, self.registry.get_sample_value('g')) - mark_process_dead(123, os.environ['prometheus_multiproc_dir']) - self.assertEqual(2, self.registry.get_sample_value('g')) +from prometheus_client import values, Histogram +from prometheus_client.core import Counter +from prometheus_client.multiprocess import mmap_collector, mmap_dict +from prometheus_client.multiprocess.mmaped_value import MmapedValue +from prometheus_client.samples import Sample +from tests.common_multiprocess_tests import CommonMultiprocessTests - def test_namespace_subsystem(self): - c1 = Counter('c', 'help', registry=None, namespace='ns', subsystem='ss') - c1.inc(1) - self.assertEqual(1, self.registry.get_sample_value('ns_ss_c_total')) - def test_counter_across_forks(self): - pid = 0 - values.ValueClass = MmapedValue(lambda: pid) - c1 = Counter('c', 'help', registry=None) - self.assertEqual(0, self.registry.get_sample_value('c_total')) - c1.inc(1) - c1.inc(1) - pid = 1 - c1.inc(1) - self.assertEqual(3, self.registry.get_sample_value('c_total')) - self.assertEqual(1, c1._value.get()) +class TestMmapMultiProcess(CommonMultiprocessTests, unittest.TestCase): + _value_class = staticmethod(MmapedValue) + _collector_class = staticmethod(mmap_collector.MmapMultiProcessCollector) + _multiprocess_backend = 'mmap' def test_initialization_detects_pid_change(self): pid = 0 - values.ValueClass = MmapedValue(lambda: pid) + values.ValueClass = self._value_class(lambda: pid) # can not inspect the files cache directly, as it's a closure, so we # check for the actual files themselves @@ -164,72 +40,10 @@ def files(): c3 = Counter('c3', 'c3', registry=None) self.assertEqual(files(), ['counter_0.db', 'counter_1.db']) - - @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") - def test_collect(self): - pid = 0 - values.ValueClass = MmapedValue(lambda: pid) - labels = dict((i, i) for i in 'abcd') - - def add_label(key, value): - l = labels.copy() - l[key] = value - return l - - c = Counter('c', 'help', labelnames=labels.keys(), registry=None) - g = Gauge('g', 'help', labelnames=labels.keys(), registry=None) - h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) - - c.labels(**labels).inc(1) - g.labels(**labels).set(1) - h.labels(**labels).observe(1) - - pid = 1 - - c.labels(**labels).inc(1) - g.labels(**labels).set(1) - h.labels(**labels).observe(5) - - metrics = dict((m.name, m) for m in self.collector.collect()) - - self.assertEqual( - metrics['c'].samples, [Sample('c_total', labels, 2.0)] - ) - metrics['g'].samples.sort(key=lambda x: x[1]['pid']) - self.assertEqual(metrics['g'].samples, [ - Sample('g', add_label('pid', '0'), 1.0), - Sample('g', add_label('pid', '1'), 1.0), - ]) - - metrics['h'].samples.sort( - key=lambda x: (x[0], float(x[1].get('le', 0))) - ) - expected_histogram = [ - Sample('h_bucket', add_label('le', '0.005'), 0.0), - Sample('h_bucket', add_label('le', '0.01'), 0.0), - Sample('h_bucket', add_label('le', '0.025'), 0.0), - Sample('h_bucket', add_label('le', '0.05'), 0.0), - Sample('h_bucket', add_label('le', '0.075'), 0.0), - Sample('h_bucket', add_label('le', '0.1'), 0.0), - Sample('h_bucket', add_label('le', '0.25'), 0.0), - Sample('h_bucket', add_label('le', '0.5'), 0.0), - Sample('h_bucket', add_label('le', '0.75'), 0.0), - Sample('h_bucket', add_label('le', '1.0'), 1.0), - Sample('h_bucket', add_label('le', '2.5'), 1.0), - Sample('h_bucket', add_label('le', '5.0'), 2.0), - Sample('h_bucket', add_label('le', '7.5'), 2.0), - Sample('h_bucket', add_label('le', '10.0'), 2.0), - Sample('h_bucket', add_label('le', '+Inf'), 2.0), - Sample('h_count', labels, 2.0), - Sample('h_sum', labels, 6.0), - ] - - self.assertEqual(metrics['h'].samples, expected_histogram) - - @unittest.skipIf(sys.version_info < (2, 7), "Test requires Python 2.7+.") + @pytest.mark.skipif(sys.version_info < (2, 7), reason="Test requires Python 2.7+.") def test_merge_no_accumulate(self): pid = 0 - values.ValueClass = MmapedValue(lambda: pid) + values.ValueClass = self._value_class(lambda: pid) labels = dict((i, i) for i in 'abcd') def add_label(key, value): @@ -270,7 +84,7 @@ def add_label(key, value): Sample('h_sum', labels, 6.0), ] - self.assertEqual(metrics['h'].samples, expected_histogram) + assert expected_histogram == metrics['h'].samples class TestMmapedDict(unittest.TestCase): @@ -304,10 +118,8 @@ def test_corruption_detected(self): self.d.write_value('abc', 42.0) # corrupt the written data self.d._m[8:16] = b'somejunk' - with self.assertRaises(RuntimeError): + with pytest.raises(RuntimeError): list(self.d.read_all_values()) def tearDown(self): os.unlink(self.tempfile) - - From 48b9e50dd7b95d810d17d3262dcd2745e2bb24dc Mon Sep 17 00:00:00 2001 From: Aarni Koskela Date: Mon, 21 Jan 2019 16:47:42 +0200 Subject: [PATCH 4/6] Break up MmapMultiProcessCollector to allow reuse Signed-off-by: Aarni Koskela --- .../multiprocess/collector_utils.py | 79 ++++++++++++++++ .../multiprocess/mmap_collector.py | 94 ++++--------------- 2 files changed, 97 insertions(+), 76 deletions(-) create mode 100644 prometheus_client/multiprocess/collector_utils.py diff --git a/prometheus_client/multiprocess/collector_utils.py b/prometheus_client/multiprocess/collector_utils.py new file mode 100644 index 00000000..e3708694 --- /dev/null +++ b/prometheus_client/multiprocess/collector_utils.py @@ -0,0 +1,79 @@ +from collections import defaultdict + +from prometheus_client.metrics_core import Metric +from prometheus_client.samples import Sample +from prometheus_client.utils import floatToGoString + + +def populate_metrics(metrics, pid, metric_name, name, labels, multiprocess_mode, type, value): + labels_key = tuple(sorted(labels.items())) + metric = metrics.get(metric_name) + if metric is None: + metric = Metric(metric_name, 'Multiprocess metric', type) + metrics[metric_name] = metric + if type == 'gauge': + metric._multiprocess_mode = multiprocess_mode + metric.add_sample(name, labels_key + (('pid', pid),), value) + else: + # The duplicates and labels are fixed in the next for. + metric.add_sample(name, labels_key, value) + + +def postprocess_metrics(metrics, accumulate=True): + for metric in metrics.values(): + metric.samples = _postprocess_metric(metric, accumulate=accumulate) + + +def _postprocess_metric(metric, accumulate=True): + samples = defaultdict(float) + buckets = {} + for s in metric.samples: + name, labels, value = s.name, s.labels, s.value + if metric.type == 'gauge': + without_pid = tuple(l for l in labels if l[0] != 'pid') + if metric._multiprocess_mode == 'min': + current = samples.setdefault((name, without_pid), value) + if value < current: + samples[(s.name, without_pid)] = value + elif metric._multiprocess_mode == 'max': + current = samples.setdefault((name, without_pid), value) + if value > current: + samples[(s.name, without_pid)] = value + elif metric._multiprocess_mode == 'livesum': + samples[(name, without_pid)] += value + else: # all/liveall + samples[(name, labels)] = value + + elif metric.type == 'histogram': + bucket = tuple(float(l[1]) for l in labels if l[0] == 'le') + if bucket: + # _bucket + without_le = tuple(l for l in labels if l[0] != 'le') + buckets.setdefault(without_le, {}) + buckets[without_le].setdefault(bucket[0], 0.0) + buckets[without_le][bucket[0]] += value + else: + # _sum/_count + samples[(s.name, labels)] += value + + else: + # Counter and Summary. + samples[(s.name, labels)] += value + # Accumulate bucket values. + if metric.type == 'histogram': + for labels, values in buckets.items(): + acc = 0.0 + for bucket, value in sorted(values.items()): + sample_key = ( + metric.name + '_bucket', + labels + (('le', floatToGoString(bucket)),), + ) + if accumulate: + acc += value + samples[sample_key] = acc + else: + samples[sample_key] = value + if accumulate: + samples[(metric.name + '_count', labels)] = acc + # Convert to correct sample format. + return [Sample(name, dict(labels), value) for (name, labels), value in samples.items()] diff --git a/prometheus_client/multiprocess/mmap_collector.py b/prometheus_client/multiprocess/mmap_collector.py index d86d457c..88887720 100644 --- a/prometheus_client/multiprocess/mmap_collector.py +++ b/prometheus_client/multiprocess/mmap_collector.py @@ -1,16 +1,13 @@ import glob import json import os -from collections import defaultdict -from prometheus_client.metrics_core import Metric +from prometheus_client.multiprocess.collector_utils import populate_metrics, postprocess_metrics from prometheus_client.multiprocess.mmap_dict import MmapedDict -from prometheus_client.samples import Sample -from prometheus_client.utils import floatToGoString class MmapMultiProcessCollector(object): - """Collector for files for multi-process mode.""" + """Collector for files for mmap multi-process mode.""" def __init__(self, registry, path): self._path = path @@ -30,80 +27,25 @@ def merge(self, files, accumulate=True): """ metrics = {} for f in files: - parts = os.path.basename(f).split('_') - typ = parts[0] + parts = os.path.basename(f).split('_') # e.g. gauge_liveall_1234.db, counter_1234.db, histogram_1234.db + typ = parts.pop(0) # grab type (remaining e.g. ['liveall', '1234.db'] or ['1234.db']) + pid = parts.pop(-1)[:-3] # grab pid off end (remaining e.g. ['liveall'] or []) + multiprocess_mode = (parts.pop(0) if parts else None) # must be the remaining multiprocess mode bit + assert not parts # make sure nothing was unread d = MmapedDict(f, read_mode=True) for key, value in d.read_all_values(): metric_name, name, labels = json.loads(key) - labels_key = tuple(sorted(labels.items())) - - metric = metrics.get(metric_name) - if metric is None: - metric = Metric(metric_name, 'Multiprocess metric', typ) - metrics[metric_name] = metric - - if typ == 'gauge': - pid = parts[2][:-3] - metric._multiprocess_mode = parts[1] - metric.add_sample(name, labels_key + (('pid', pid), ), value) - else: - # The duplicates and labels are fixed in the next for. - metric.add_sample(name, labels_key, value) + populate_metrics( + metrics, + pid=pid, + metric_name=metric_name, + name=name, + labels=labels, + multiprocess_mode=multiprocess_mode, + type=typ, + value=value, + ) d.close() - for metric in metrics.values(): - samples = defaultdict(float) - buckets = {} - for s in metric.samples: - name, labels, value = s.name, s.labels, s.value - if metric.type == 'gauge': - without_pid = tuple(l for l in labels if l[0] != 'pid') - if metric._multiprocess_mode == 'min': - current = samples.setdefault((name, without_pid), value) - if value < current: - samples[(s.name, without_pid)] = value - elif metric._multiprocess_mode == 'max': - current = samples.setdefault((name, without_pid), value) - if value > current: - samples[(s.name, without_pid)] = value - elif metric._multiprocess_mode == 'livesum': - samples[(name, without_pid)] += value - else: # all/liveall - samples[(name, labels)] = value - - elif metric.type == 'histogram': - bucket = tuple(float(l[1]) for l in labels if l[0] == 'le') - if bucket: - # _bucket - without_le = tuple(l for l in labels if l[0] != 'le') - buckets.setdefault(without_le, {}) - buckets[without_le].setdefault(bucket[0], 0.0) - buckets[without_le][bucket[0]] += value - else: - # _sum/_count - samples[(s.name, labels)] += value - - else: - # Counter and Summary. - samples[(s.name, labels)] += value - - # Accumulate bucket values. - if metric.type == 'histogram': - for labels, values in buckets.items(): - acc = 0.0 - for bucket, value in sorted(values.items()): - sample_key = ( - metric.name + '_bucket', - labels + (('le', floatToGoString(bucket)), ), - ) - if accumulate: - acc += value - samples[sample_key] = acc - else: - samples[sample_key] = value - if accumulate: - samples[(metric.name + '_count', labels)] = acc - - # Convert to correct sample format. - metric.samples = [Sample(name, dict(labels), value) for (name, labels), value in samples.items()] + postprocess_metrics(metrics, accumulate) return metrics.values() From 570df71bc8bf06436d4fb885716b3ec186e21e5b Mon Sep 17 00:00:00 2001 From: Aarni Koskela Date: Mon, 21 Jan 2019 16:54:06 +0200 Subject: [PATCH 5/6] Add SQLite backend for multiprocess values Signed-off-by: Aarni Koskela --- prometheus_client/multiprocess/__init__.py | 9 ++ prometheus_client/multiprocess/sqlite.py | 133 +++++++++++++++++++++ tests/test_multiprocess_sqlite.py | 12 ++ 3 files changed, 154 insertions(+) create mode 100644 prometheus_client/multiprocess/sqlite.py create mode 100644 tests/test_multiprocess_sqlite.py diff --git a/prometheus_client/multiprocess/__init__.py b/prometheus_client/multiprocess/__init__.py index 8388f97d..0b52b7af 100644 --- a/prometheus_client/multiprocess/__init__.py +++ b/prometheus_client/multiprocess/__init__.py @@ -14,6 +14,9 @@ def MultiProcessCollector(registry, path=None): if multiprocess_backend == 'mmap': from prometheus_client.multiprocess.mmap_collector import MmapMultiProcessCollector return MmapMultiProcessCollector(registry=registry, path=path) + elif multiprocess_backend == 'sqlite': + from prometheus_client.multiprocess.sqlite import SqliteMultiProcessCollector + return SqliteMultiProcessCollector(registry=registry, path=path) else: raise NotImplementedError('unknown multiprocess backend %s' % multiprocess_backend) @@ -22,6 +25,9 @@ def get_multiprocess_value_class(): if multiprocess_backend == 'mmap': from prometheus_client.multiprocess.mmaped_value import MmapedValue return MmapedValue() + elif multiprocess_backend == 'sqlite': + from prometheus_client.multiprocess.sqlite import SqliteValue + return SqliteValue() else: raise NotImplementedError('unknown multiprocess backend %s' % multiprocess_backend) @@ -33,5 +39,8 @@ def mark_process_dead(pid, path=None): if multiprocess_backend == 'mmap': from prometheus_client.multiprocess.mmap_utils import mmap_cleanup mmap_cleanup(path, pid) + elif multiprocess_backend == 'sqlite': + from prometheus_client.multiprocess.sqlite import sqlite_cleanup + sqlite_cleanup(path, pid) else: raise NotImplementedError('unknown multiprocess backend %s' % multiprocess_backend) diff --git a/prometheus_client/multiprocess/sqlite.py b/prometheus_client/multiprocess/sqlite.py new file mode 100644 index 00000000..a5fd411f --- /dev/null +++ b/prometheus_client/multiprocess/sqlite.py @@ -0,0 +1,133 @@ +from prometheus_client.multiprocess.collector_utils import populate_metrics, postprocess_metrics +import contextlib +import json + +import base64 +import hashlib +import os +import sqlite3 + + +def sqlite_key(metric_name, name, labelnames, labelvalues, type, multiprocess_mode): + """Format a key for use in the SQLite file.""" + # ensure labels are in consistent order for identity + labels = dict(zip(labelnames, labelvalues)) + return json.dumps([metric_name, name, labels, type, multiprocess_mode], sort_keys=True) + + +def get_database(directory): + db_filename = os.path.join(directory, 'prometheus_python.sqlite3') + db = sqlite3.connect(db_filename, isolation_level=None) + db.executescript(""" + PRAGMA journal_mode = OFF; + CREATE TABLE IF NOT EXISTS prom_values ( + pid TEXT, + muid TEXT, + m_key TEXT, + multiprocess_mode TEXT NULL, + value REAL + ); + CREATE UNIQUE INDEX IF NOT EXISTS ix_value ON prom_values (pid, muid); + """) + return db + + +def sqlite_cleanup(path, pid): + with contextlib.closing(get_database(path)) as db: + res = db.execute( + 'DELETE FROM prom_values WHERE pid = ? AND multiprocess_mode IN ("livesum", "liveall")', + (str(pid),), + ) + + +class SqliteMultiProcessCollector(object): + """Collector for files for SQlite multi-process mode.""" + + def __init__(self, registry, path): + self._path = path + if registry: + registry.register(self) + + def collect(self): + metrics = {} + # type db: sqlite3.Connection + with contextlib.closing(self.get_database()) as db: + for pid, key, value in db.execute('SELECT pid, m_key, value FROM prom_values'): + metric_name, name, labels, type, multiprocess_mode = json.loads(key) + populate_metrics( + metrics, + pid=pid, + metric_name=metric_name, + name=name, + labels=labels, + multiprocess_mode=multiprocess_mode, + type=type, + value=value, + ) + + postprocess_metrics(metrics, accumulate=True) + return metrics.values() + + def get_database(self): + return get_database(self._path) + + +class _SqliteValue(object): + _multiprocess = True + + # Filled in by subclasses: + database = None # type: sqlite3.Connection + pid_func = None # type: function + + def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', **kwargs): + self.type = typ + self.metric_name = metric_name + self.name = name + self.labelnames = labelnames + self.labelvalues = labelvalues + self.multiprocess_mode = multiprocess_mode + self._key = sqlite_key( + metric_name=self.metric_name, + name=self.name, + labelnames=self.labelnames, + labelvalues=self.labelvalues, + type=self.type, + multiprocess_mode=multiprocess_mode, + ) + self._muid = base64.b64encode(hashlib.sha256(self._key.encode('UTF-8')).digest()).decode() + self.set(0) + + def inc(self, amount): + pid = self.pid_func() + self.database.execute( + 'INSERT INTO prom_values (pid, muid, m_key, multiprocess_mode, value) VALUES (?, ?, ?, ?, ?) ' + 'ON CONFLICT (pid, muid) DO UPDATE SET value = value + ?', + (pid, self._muid, self._key, self.multiprocess_mode, amount, amount) + ) + + def set(self, value): + pid = self.pid_func() + self.database.execute( + 'INSERT INTO prom_values (pid, muid, m_key, multiprocess_mode, value) VALUES (?, ?, ?, ?, ?) ' + 'ON CONFLICT (pid, muid) DO UPDATE SET value = ?', + (pid, self._muid, self._key, self.multiprocess_mode, value, value) + ) + + def get(self): + pid = self.pid_func() + result = self.database.execute( + 'SELECT value FROM prom_values WHERE pid = ? AND muid = ? LIMIT 1', + (pid, self._muid), + ) + row = result.fetchone() + return (row[0] if row else 0) + + +def SqliteValue(_pidFunc=os.getpid, directory=None): + directory = (directory or os.environ['prometheus_multiproc_dir']) + + class SqliteValue(_SqliteValue): + database = get_database(directory) + pid_func = staticmethod(_pidFunc) + + return SqliteValue diff --git a/tests/test_multiprocess_sqlite.py b/tests/test_multiprocess_sqlite.py new file mode 100644 index 00000000..487c05a9 --- /dev/null +++ b/tests/test_multiprocess_sqlite.py @@ -0,0 +1,12 @@ +from __future__ import unicode_literals + +import unittest + +from prometheus_client.multiprocess.sqlite import SqliteMultiProcessCollector, SqliteValue +from tests.common_multiprocess_tests import CommonMultiprocessTests + + +class TestSqliteMultiProcess(CommonMultiprocessTests, unittest.TestCase): + _value_class = staticmethod(SqliteValue) + _collector_class = staticmethod(SqliteMultiProcessCollector) + _multiprocess_backend = 'sqlite' From a89425f105ba57a12793ce0f6592292e7560abf7 Mon Sep 17 00:00:00 2001 From: Aarni Koskela Date: Mon, 21 Jan 2019 18:22:28 +0200 Subject: [PATCH 6/6] Support SQLite versions without upsert Signed-off-by: Aarni Koskela --- prometheus_client/multiprocess/sqlite.py | 41 ++++++++++++++++++------ 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/prometheus_client/multiprocess/sqlite.py b/prometheus_client/multiprocess/sqlite.py index a5fd411f..340154b7 100644 --- a/prometheus_client/multiprocess/sqlite.py +++ b/prometheus_client/multiprocess/sqlite.py @@ -7,6 +7,9 @@ import os import sqlite3 +# "UPSERT syntax was added to SQLite with version 3.24.0 (2018-06-04)." +has_upsert = (sqlite3.sqlite_version_info[:2] >= (3, 24)) + def sqlite_key(metric_name, name, labelnames, labelvalues, type, multiprocess_mode): """Format a key for use in the SQLite file.""" @@ -99,19 +102,37 @@ def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess def inc(self, amount): pid = self.pid_func() - self.database.execute( - 'INSERT INTO prom_values (pid, muid, m_key, multiprocess_mode, value) VALUES (?, ?, ?, ?, ?) ' - 'ON CONFLICT (pid, muid) DO UPDATE SET value = value + ?', - (pid, self._muid, self._key, self.multiprocess_mode, amount, amount) - ) + if has_upsert: + self.database.execute( + 'INSERT INTO prom_values (pid, muid, m_key, multiprocess_mode, value) ' + 'VALUES (?, ?, ?, ?, ?) ' + 'ON CONFLICT (pid, muid) DO UPDATE SET value = value + ?', + (pid, self._muid, self._key, self.multiprocess_mode, amount, amount) + ) + else: + # This is not 100% atomic. + cur = self.database.execute( + 'UPDATE prom_values SET value = value + ? WHERE pid = ? AND muid = ?', + (amount, pid, self._muid) + ) + if not cur.rowcount: + self.set(amount) def set(self, value): pid = self.pid_func() - self.database.execute( - 'INSERT INTO prom_values (pid, muid, m_key, multiprocess_mode, value) VALUES (?, ?, ?, ?, ?) ' - 'ON CONFLICT (pid, muid) DO UPDATE SET value = ?', - (pid, self._muid, self._key, self.multiprocess_mode, value, value) - ) + if has_upsert: + self.database.execute( + 'INSERT INTO prom_values (pid, muid, m_key, multiprocess_mode, value) ' + 'VALUES (?, ?, ?, ?, ?) ' + 'ON CONFLICT (pid, muid) DO UPDATE SET value = ?', + (pid, self._muid, self._key, self.multiprocess_mode, value, value) + ) + else: + self.database.execute( + 'INSERT OR REPLACE INTO prom_values (pid, muid, m_key, multiprocess_mode, value) ' + 'VALUES (?, ?, ?, ?, ?)', + (pid, self._muid, self._key, self.multiprocess_mode, value) + ) def get(self): pid = self.pid_func()