diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml new file mode 100644 index 00000000..1de9b40b --- /dev/null +++ b/.github/codeql/codeql-config.yml @@ -0,0 +1,4 @@ +name: "CodeQL config" + +queries: + - uses: security-extended \ No newline at end of file diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml new file mode 100644 index 00000000..a1988a3a --- /dev/null +++ b/.github/workflows/codeql-analysis.yml @@ -0,0 +1,53 @@ +name: "Code scanning - action" + +on: + push: + branches: [master, ] + schedule: + - cron: '0 8 * * 2' + +jobs: + CodeQL-Build: + + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v2 + with: + # We must fetch at least the immediate parents so that if this is + # a pull request then we can checkout the head. + fetch-depth: 2 + + # If this run was triggered by a pull request event, then checkout + # the head of the pull request instead of the merge commit. + - run: git checkout HEAD^2 + if: ${{ github.event_name == 'pull_request' }} + + # Initializes the CodeQL tools for scanning. + - name: Initialize CodeQL + uses: github/codeql-action/init@v1 + # Override language selection by uncommenting this and choosing your languages + # with: + # languages: go, javascript, csharp, python, cpp, java + with: + config-file: ./.github/codeql/codeql-config.yml + + # Autobuild attempts to build any compiled languages (C/C++, C#, or Java). + # If this step fails, then you should remove it and run the build manually (see below) + - name: Autobuild + uses: github/codeql-action/autobuild@v1 + + # ℹī¸ Command-line programs to run using the OS shell. + # 📚 https://git.io/JvXDl + + # ✏ī¸ If the Autobuild fails above, remove it and uncomment the following three lines + # and modify them (or add more) to build your code if your project + # uses a compiled language + + #- run: | + # make bootstrap + # make release + + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v1 \ No newline at end of file diff --git a/.gitignore b/.gitignore index 0ba244f6..f4472290 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,6 @@ dist .coverage .tox .*cache -htmlcov \ No newline at end of file +htmlcov +.venv +tmp diff --git a/README.md b/README.md index 46671b76..8c7c0735 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,21 @@ +# Rationale for the Fork +We've forked the original prometheus python client for a few reasons, all stemming from multiprocessing handling + +## Context +Postal-main uses a pretty standard gunicorn deployment (pre-fork web server). Glossing over many details: a gunicorn master process will fork worker processes to allow handling web requests in parallel. These workers process a few hundred requests, then exit, getting replaced by another forked worker. The use of processes, instead of threads, makes aggregating metrics from the workers less straightforward. What the mainline repo does, and what we do, is to use the filesystem to store the metrics for each of these workers. When prometheus scrapes the scrape endpoint, the client will read all the relevant metrics files, aggregate the results, then serve the metrics. + +Likewise, our celery pods use `--pool=prefork`, and metrics need to be collected the same way + +## What we do differently +In a multiprocess setup, in which we treat multiple OS processes as one logical process (i.e. we are interested in the metrics for a pod, not each gunicorn worker in the pod), we need aggregate the metrics somehow. For counters and histograms, this is straightforward; we just sum up everything. For gauges, there are multiple strategies for aggregating these metrics, all equally valid. We might want to that the max value of all the process gauges for a high-water-mark, or we might want to, for a batch job that is fanned out to multiple workers, take a sum of all the gauges for live worker processes to track progress on that batch. + +Many of our celery tasks have a pattern of querying the database to find entries which need to processed in some way, e.g. couriers which need to be paid out. Because of how we process this data, we needed a gauge aggregation strategy which just takes the latest gauge value, discarding the rest (search for "multiprocess_mode" below + +Metrics files are identified by a pid. Since processes are constantly forked and exiting, this will, at best, generate a lot of metrics files, which all need to be opened for a scrape, and at worst, could have pid collisions. This fork runs a thread which goes and cleans up metrics files generated by exited processes, and merges them into an archive file + +# *(Mostly) Original Readme Below* +--- + # Prometheus Python Client The official Python 2 and 3 client for [Prometheus](http://prometheus.io). @@ -235,8 +253,8 @@ ProcessCollector(namespace='mydaemon', pid=lambda: open('/var/run/daemon.pid').r ### Platform Collector The client also automatically exports some metadata about Python. If using Jython, -metadata about the JVM in use is also included. This information is available as -labels on the `python_info` metric. The value of the metric is 1, since it is the +metadata about the JVM in use is also included. This information is available as +labels on the `python_info` metric. The value of the metric is 1, since it is the labels that carry information. ## Exporting @@ -459,7 +477,7 @@ implement a proper `describe`, or if that's not practical have `describe` return an empty list. -## Multiprocess Mode (Gunicorn) +## Multiprocess Mode Prometheus client libaries presume a threaded model, where metrics are shared across workers. This doesn't work so well for languages such as Python where @@ -476,20 +494,71 @@ This comes with a number of limitations: There's several steps to getting this working: -**One**: Gunicorn deployment +**One**: Deployment The `prometheus_multiproc_dir` environment variable must be set to a directory -that the client library can use for metrics. This directory must be wiped -between Gunicorn runs (before startup is recommended). +that the client library can use to share metric databases between processes. In production it is recommended +to use a tmpfs volume (e.g. `/tmp/prometheus`) for this directory so that the exporter doesn't interfere +disk IO. + +Application workers write to metric databases in this directory. The exporter +process reads from it and merges dead application worker metric databases. + +### Option A: Integrating with an existing Gunicorn/WSGI application: + +Add the following to your Gunicorn config file: -Put the following in the config file: ```python -from prometheus_client import multiprocess +from prometheus_client import multiprocess_exporter -def child_exit(server, worker): - multiprocess.mark_process_dead(worker.pid) +def on_starting(server): + multiprocess.start_archiver_thread() ``` +Add the Prometheus Exporter WSGI handler to your existing WSGI handler + +```python +from prometheus_client import multiprocess_exporter + +.... +def app(environ, start_response): + if environ.get("PATH_INFO") == "/metrics": + return multiprocess_exporter.app(environ, start_response) + else: + return ... +``` + +### Option B (Celery and other applications): Run a sidecar Gunicorn process to export the metrics + +In the same filesystem as your other Python application, start an exporter sidecar + +```shell +#!/bin/bash + +export prometheus_multiproc_dir=/tmp/prometheus # some dir +# Start the sidecar exporter: +( + set -eu + mkdir -p ${prometheus_multiproc_dir} + exec gunicorn \ + --config python:prometheus_client.prometheus_exporter \ + --preload \ + --workers 1 \ + --threads 10 \ + --bind 0.0.0.0:9500 \ + prometheus_client.prometheus_exporter:app +) & + +# Start the application: +celery ... + +``` + +This will export the metrics at `http://127.0.0.1:9500` + +Only one exporter process should run per filesystem, prometheus_multiproc_dir. + + **Two**: Inside the application ```python from prometheus_client import multiprocess @@ -522,6 +591,7 @@ Gauges have several modes they can run in, which can be selected with the `multiprocess_mode` parameter. - 'all': Default. Return a timeseries per process alive or dead. +- 'latest': Default. Return the most recent gauge update value. - 'liveall': Return a timeseries per process that is still alive. - 'livesum': Return a single timeseries that is the sum of the values of alive processes. - 'max': Return a single timeseries that is the maximum of the values of all processes, alive or dead. diff --git a/prometheus_client/metrics.py b/prometheus_client/metrics.py index b7c5e5a4..ca50d025 100644 --- a/prometheus_client/metrics.py +++ b/prometheus_client/metrics.py @@ -65,8 +65,8 @@ def describe(self): def collect(self): metric = self._get_metric() - for suffix, labels, value in self._samples(): - metric.add_sample(self._name + suffix, labels, value) + for suffix, labels, value, timestamp in self._samples(): + metric.add_sample(self._name + suffix, labels, value, timestamp=timestamp) return [metric] def __init__(self, @@ -179,8 +179,8 @@ def _multi_samples(self): metrics = self._metrics.copy() for labels, metric in metrics.items(): series_labels = list(zip(self._labelnames, labels)) - for suffix, sample_labels, value in metric._samples(): - yield (suffix, dict(series_labels + list(sample_labels.items())), value) + for suffix, sample_labels, value, timestamp in metric._samples(): + yield (suffix, dict(series_labels + list(sample_labels.items())), value, timestamp) def _child_samples(self): # pragma: no cover raise NotImplementedError('_child_samples() must be implemented by %r' % self) @@ -250,8 +250,8 @@ def count_exceptions(self, exception=Exception): def _child_samples(self): return ( - ('_total', {}, self._value.get()), - ('_created', {}, self._created), + ('_total', {}, self._value.get(), None), + ('_created', {}, self._created, None), ) @@ -293,7 +293,15 @@ def f(): d.set_function(lambda: len(my_dict)) """ _type = 'gauge' - _MULTIPROC_MODES = frozenset(('min', 'max', 'livesum', 'liveall', 'all')) + ALL = "all" + LATEST = "latest" + LIVEALL = "liveall" + LIVESUM = "livesum" + MAX = "max" + MIN = "min" + + + _MULTIPROC_MODES = frozenset((MIN, MAX, LATEST, LIVESUM, LIVEALL, ALL)) def __init__(self, name, @@ -304,9 +312,10 @@ def __init__(self, unit='', registry=REGISTRY, labelvalues=None, - multiprocess_mode='all', + multiprocess_mode='latest', ): self._multiprocess_mode = multiprocess_mode + self._f = None if multiprocess_mode not in self._MULTIPROC_MODES: raise ValueError('Invalid multiprocess mode: ' + multiprocess_mode) super(Gauge, self).__init__( @@ -327,21 +336,27 @@ def _metric_init(self): multiprocess_mode=self._multiprocess_mode ) - def inc(self, amount=1): + def _current_time(self, timestamp): + """Get the current unixtime for the LATEST multiprocess_mode""" + if not timestamp and self._multiprocess_mode == self.LATEST: + timestamp = time.time() + return timestamp + + def inc(self, amount=1, timestamp=None): """Increment gauge by the given amount.""" - self._value.inc(amount) + self._value.inc(amount, timestamp=self._current_time(timestamp)) - def dec(self, amount=1): + def dec(self, amount=1, timestamp=None): """Decrement gauge by the given amount.""" - self._value.inc(-amount) + self._value.inc(-amount, timestamp=self._current_time(timestamp)) - def set(self, value): + def set(self, value, timestamp=None): """Set gauge to the given value.""" - self._value.set(float(value)) + self._value.set(float(value), timestamp=self._current_time(timestamp)) - def set_to_current_time(self): + def set_to_current_time(self, timestamp=None): """Set gauge to the current unixtime.""" - self.set(time.time()) + self.set(time.time(), timestamp=self._current_time(timestamp)) def track_inprogress(self): """Track inprogress blocks of code or functions. @@ -367,12 +382,12 @@ def set_function(self, f): """ def samples(self): - return (('', {}, float(f())),) + return (('', {}, float(f()), None),) self._child_samples = types.MethodType(samples, self) def _child_samples(self): - return (('', {}, self._value.get()),) + return (('', {}, self._value.get(), self._value.timestamp()),) class Summary(MetricWrapperBase): @@ -428,9 +443,9 @@ def time(self): def _child_samples(self): return ( - ('_count', {}, self._count.get()), - ('_sum', {}, self._sum.get()), - ('_created', {}, self._created)) + ('_count', {}, self._count.get(), None), + ('_sum', {}, self._sum.get(), None), + ('_created', {}, self._created, None)) class Histogram(MetricWrapperBase): @@ -542,10 +557,10 @@ def _child_samples(self): acc = 0 for i, bound in enumerate(self._upper_bounds): acc += self._buckets[i].get() - samples.append(('_bucket', {'le': floatToGoString(bound)}, acc)) - samples.append(('_count', {}, acc)) - samples.append(('_sum', {}, self._sum.get())) - samples.append(('_created', {}, self._created)) + samples.append(('_bucket', {'le': floatToGoString(bound)}, acc, None)) + samples.append(('_count', {}, acc, None)) + samples.append(('_sum', {}, self._sum.get(), None)) + samples.append(('_created', {}, self._created, None)) return tuple(samples) @@ -582,7 +597,7 @@ def info(self, val): def _child_samples(self): with self._lock: - return (('_info', self._value, 1.0,),) + return (('_info', self._value, 1.0, None),) class Enum(MetricWrapperBase): @@ -639,7 +654,7 @@ def state(self, state): def _child_samples(self): with self._lock: return [ - ('', {self._name: s}, 1 if i == self._value else 0,) + ('', {self._name: s}, 1 if i == self._value else 0, None) for i, s in enumerate(self._states) ] diff --git a/prometheus_client/mmap_dict.py b/prometheus_client/mmap_dict.py index 679597fa..f132263e 100644 --- a/prometheus_client/mmap_dict.py +++ b/prometheus_client/mmap_dict.py @@ -5,17 +5,15 @@ _INITIAL_MMAP_SIZE = 1 << 20 _pack_integer_func = struct.Struct(b'i').pack -_pack_double_func = struct.Struct(b'd').pack +_value_timestamp = struct.Struct(b'dd') _unpack_integer = struct.Struct(b'i').unpack_from -_unpack_double = struct.Struct(b'd').unpack_from # struct.pack_into has atomicity issues because it will temporarily write 0 into # the mmap, resulting in false reads to 0 when experiencing a lot of writes. # Using direct assignment solves this issue. - -def _pack_double(data, pos, value): - data[pos:pos + 8] = _pack_double_func(value) +def _pack_value_timestamp(data, pos, value, timestamp): + data[pos:pos + _value_timestamp.size] = _value_timestamp.pack(value, timestamp) def _pack_integer(data, pos, value): @@ -29,7 +27,7 @@ class MmapedDict(object): Then 4 bytes of padding. There's then a number of entries, consisting of a 4 byte int which is the size of the next field, a utf-8 encoded string key, padding to a 8 byte - alignment, and then a 8 byte float which is the value. + alignment, a 8 byte float which is the value and then an 8 byte timestamp (seconds). Not thread safe. """ @@ -50,7 +48,7 @@ def __init__(self, filename, read_mode=False): _pack_integer(self._m, 0, self._used) else: if not read_mode: - for key, _, pos in self._read_all_values(): + for key, _, _, pos in self._read_all_values(): self._positions[key] = pos def _init_value(self, key): @@ -58,7 +56,7 @@ def _init_value(self, key): encoded = key.encode('utf-8') # Pad to be 8-byte aligned. padded = encoded + (b' ' * (8 - (len(encoded) + 4) % 8)) - value = struct.pack('i{0}sd'.format(len(padded)).encode(), len(encoded), padded, 0.0) + value = struct.pack('i{0}sdd'.format(len(padded)).encode(), len(encoded), padded, 0.0, 0.0) while self._used + len(value) > self._capacity: self._capacity *= 2 self._f.truncate(self._capacity) @@ -68,10 +66,10 @@ def _init_value(self, key): # Update how much space we've used. self._used += len(value) _pack_integer(self._m, 0, self._used) - self._positions[key] = self._used - 8 + self._positions[key] = self._used - _value_timestamp.size def _read_all_values(self): - """Yield (key, value, pos). No locking is performed.""" + """Yield (key, value, timestamp, pos). No locking is performed.""" pos = 8 @@ -91,28 +89,32 @@ def _read_all_values(self): encoded = unpack_from(('%ss' % encoded_len).encode(), data, pos)[0] padded_len = encoded_len + (8 - (encoded_len + 4) % 8) pos += padded_len - value = _unpack_double(data, pos)[0] - yield encoded.decode('utf-8'), value, pos - pos += 8 + value, timestamp = _value_timestamp.unpack_from(data, pos) + yield encoded.decode('utf-8'), value, _from_timestamp_float(timestamp), pos + pos += _value_timestamp.size def read_all_values(self): """Yield (key, value, pos). No locking is performed.""" - for k, v, _ in self._read_all_values(): - yield k, v + for k, v, ts, _ in self._read_all_values(): + yield k, v, ts - def read_value(self, key): + def read_value_timestamp(self, key): if key not in self._positions: self._init_value(key) pos = self._positions[key] # We assume that reading from an 8 byte aligned value is atomic - return _unpack_double(self._m, pos)[0] + val, ts = _value_timestamp.unpack_from(self._m, pos) + return val, _from_timestamp_float(ts) - def write_value(self, key, value): + def read_value(self, key): + return self.read_value_timestamp(key)[0] + + def write_value(self, key, value, timestamp=None): if key not in self._positions: self._init_value(key) pos = self._positions[key] # We assume that writing to an 8 byte aligned value is atomic - _pack_double(self._m, pos, value) + _pack_value_timestamp(self._m, pos, value, _to_timestamp_float(timestamp)) def close(self): if self._f: @@ -127,3 +129,25 @@ def mmap_key(metric_name, name, labelnames, labelvalues): # ensure labels are in consistent order for identity labels = dict(zip(labelnames, labelvalues)) return json.dumps([metric_name, name, labels], sort_keys=True) + + +def _from_timestamp_float(timestamp): + """Convert timestamp from a pure floating point value + + inf is decoded as None + """ + if timestamp == float('inf'): + return None + else: + return timestamp + + +def _to_timestamp_float(timestamp): + """Convert timestamp to a pure floating point value + + None is encoded as inf + """ + if timestamp is None: + return float('inf') + else: + return float(timestamp) diff --git a/prometheus_client/multiprocess.py b/prometheus_client/multiprocess.py index e34ced03..619dba09 100644 --- a/prometheus_client/multiprocess.py +++ b/prometheus_client/multiprocess.py @@ -3,14 +3,76 @@ from __future__ import unicode_literals from collections import defaultdict +from contextlib import contextmanager +import errno +from fcntl import flock, LOCK_EX, LOCK_NB, LOCK_SH, LOCK_UN import glob import json +import logging import os +import re +import shutil +import tempfile +from threading import RLock +import time -from .metrics_core import Metric -from .mmap_dict import MmapedDict +from .metrics import Counter, Gauge, Histogram +from .metrics_core import GaugeMetricFamily, Metric +from .mmap_dict import mmap_key, MmapedDict from .samples import Sample from .utils import floatToGoString +from .vendor import six + +PROMETHEUS_MULTIPROC_DIR = "prometheus_multiproc_dir" +_db_pattern = re.compile(r"(\w+)_(\d+)\.db") + + +class MetricsCache(object): + """ + A singleton which, in conjunction with the archiver thread, maintains + the last collected set of metrics. + + The MetricsCache can also be registered as a collector, providing + information about the duration of the last archive attempt + """ + def __init__(self): + self.metrics = [] + self.last_archive_duration = 0 + self.lock = RLock() + + def retrieve_metrics(self): + with self.lock: + return self.metrics + + def write_metrics(self, metrics, time_elapsed=None): + with self.lock: + self.last_archive_duration = time_elapsed + self.metrics = metrics + + def collect(self): + with self.lock: + return [GaugeMetricFamily( + "prom_client_archive_duration_seconds", + "Time taken to collect the latest batch of metrics", + value=self.last_archive_duration), ] + + +_metrics_cache = MetricsCache() + + +class InMemoryCollector(object): + """ + A Collector which simply serves metrics collected by the archiver + and stored in the MetricsCache + """ + + def __init__(self, registry): + if registry: + registry.register(self) + registry.register(_metrics_cache) + + def collect(self): + return _metrics_cache.retrieve_metrics() class MultiProcessCollector(object): @@ -25,104 +87,330 @@ def __init__(self, registry, path=None): if registry: registry.register(self) - @staticmethod - def merge(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 = {} + def collect(self, blocking=True): + # blocking=False is used for testing purposes + lock_type = LOCK_SH if blocking else LOCK_SH | LOCK_NB + with advisory_lock(lock_type): + files = glob.glob(os.path.join(self._path, '*.db')) + return merge(files, accumulate=True) + + +def merge(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 = load_metrics_from_files(files) + + for metric in six.itervalues(metrics): + # Handle the Gauge "latest" multiprocess mode type: + if metric.type == Gauge._type and metric._multiprocess_mode == Gauge.LATEST: + s = max(metric.samples, key=lambda i: i.timestamp) + # Group samples by name, labels: + grouped_samples = defaultdict(list) 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 + labels = dict(s.labels) + if "pid" in labels: + del labels["pid"] + grouped_samples[s.name, tuple(sorted(labels.items()))].append(s) + metric.samples = [] + for (name, labels), sample_group in six.iteritems(grouped_samples): + s = max(sample_group, key=lambda i: i.timestamp) + metric.samples.append(Sample(name, + dict(labels), + value=s.value, + timestamp=s.timestamp)) + continue + samples = defaultdict(float) + buckets = {} + for s in metric.samples: + name, labels, value = s.name, s.labels, s.value + if metric.type == Gauge._type: + without_pid = tuple(l for l in labels if l[0] != 'pid') + if metric._multiprocess_mode == Gauge.MIN: + current = samples.setdefault((name, without_pid), value) + if value < current: + samples[(s.name, without_pid)] = value + elif metric._multiprocess_mode == Gauge.MAX: + current = samples.setdefault((name, without_pid), value) + if value > current: + samples[(s.name, without_pid)] = value + elif metric._multiprocess_mode == Gauge.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: - # Counter and Summary. + # _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 + # 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: - samples[(metric.name + '_count', labels)] = acc + 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() - # Convert to correct sample format. - metric.samples = [Sample(name_, dict(labels), value) for (name_, labels), value in samples.items()] - return metrics.values() - def collect(self): - files = glob.glob(os.path.join(self._path, '*.db')) - return self.merge(files, accumulate=True) +def load_metrics_from_files(files): + # TODO: read from + metrics = {} + for f in files: + parts = os.path.splitext(os.path.basename(f))[0].split('_') + typ = parts[0] + multiprocess_mode = parts[1] if typ == Gauge._type else None + pid = parts[2] if multiprocess_mode and len(parts) > 2 else None + try: + d = MmapedDict(f, read_mode=True) + except EnvironmentError: + # The liveall and livesum gauge metrics + # are deleted when the gunicorn/celery worker process dies + # (mark_process_dead and, in postal-main, boot.gunicornconf.child_exit). + # Since these are deleted without acquiring a lock, they may + # not be present in between collecting the metrics files and + # merging them, resulting in a FileNotFoundError/IOError. + # However, since these gauges only care about live processes, + # we wouldn't merge them anyway. + # + # Additionally, we have a single thread which will collect + # metrics files from dead workers, and merge them into a set of + # archive files at regular interviews (see + # multiprocess_exporter). This operation is protected by a + # mutex, ensuring that no collectors are run during cleanup. We + # must do so because other metrics are sensitive to partial + # collection; prometheus counters cannot be decremented, as + # prometheus will assume that, in the time since the last scrape, + # the counter reset to 0 and incremented back up to the + # collected value, manifesting as a huge rate spike + if typ == 'gauge' and parts[1] in (Gauge.LIVESUM, Gauge.LIVEALL): + continue + raise + + for key, value, timestamp in d.read_all_values(): + metric_name, name, labels = json.loads(key) + if pid: + labels["pid"] = pid + 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 multiprocess_mode: + metric._multiprocess_mode = multiprocess_mode + metric.add_sample(name, labels_key, value, timestamp=timestamp) + d.close() + return metrics 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) + path = _multiproc_dir() if path is None else path + _remove_livesum_dbs(pid, path=path) + + +def _remove_livesum_dbs(pid, path): + for gauge_type in [Gauge.LIVESUM, Gauge.LIVEALL]: + _safe_remove("{}/gauge_{}_{}.db".format(path, gauge_type, pid)) + + +def _multiproc_dir(): + return os.environ[PROMETHEUS_MULTIPROC_DIR] + + +def _get_archive_paths(prom_dir=None): + prom_dir = _multiproc_dir() if prom_dir is None else prom_dir + merged_paths = { + (Histogram._type, None): "histogram.db", + (Counter._type, None): "counter.db", + (Gauge._type, Gauge.LATEST): "gauge_{}.db".format(Gauge.LATEST), + (Gauge._type, Gauge.MAX): "gauge_{}.db".format(Gauge.MAX), + (Gauge._type, Gauge.MIN): "gauge_{}.db".format(Gauge.MIN), + } + merged_paths = { + k: os.path.join(prom_dir, f) for k, f in six.iteritems(merged_paths) + } + return merged_paths + + +def cleanup_process(pid, prom_dir=None): + """Aggregate dead worker's metrics into a single archive file.""" + prom_dir = _multiproc_dir() if prom_dir is None else prom_dir + + merged_paths = _get_archive_paths(prom_dir) + worker_paths = [ + "counter_{}.db".format(pid), + "gauge_{}_{}.db".format(Gauge.LATEST, pid), + "gauge_{}_{}.db".format(Gauge.MAX, pid), + "gauge_{}_{}.db".format(Gauge.MIN, pid), + "histogram_{}.db".format(pid), + ] + worker_paths = (os.path.join(prom_dir, f) for f in worker_paths) + worker_paths = list(filter(os.path.exists, worker_paths)) + if worker_paths: + all_paths = worker_paths + list(filter(os.path.exists, merged_paths.values())) + metrics = merge(all_paths, accumulate=False) + _write_metrics(metrics, merged_paths) + for worker_path in worker_paths: + _safe_remove(worker_path) + _remove_livesum_dbs(pid, path=prom_dir) + + +def _safe_remove(p): + try: + os.unlink(p) + except OSError as e: + if e.errno != errno.ENOENT: + raise + + +def _write_metrics(metrics, metric_type_to_dst_path): + # TODO: use of mktemp is discouraged + mmaped_dicts = defaultdict(lambda: MmapedDict(tempfile.mktemp())) + for metric in metrics: + if metric.type not in [Histogram._type, Counter._type, Gauge._type]: + continue + mode = None + if metric.type == Gauge._type: + mode = metric._multiprocess_mode + if mode not in [Gauge.MIN, Gauge.MAX, Gauge.LATEST]: + continue + sink = mmaped_dicts[metric.type, mode] + + for sample in metric.samples: + # prometheus_client 0.4+ adds extra fields + key = mmap_key( + metric.name, + sample.name, + tuple(sample.labels), + tuple(sample.labels.values()), + ) + sink.write_value(key, sample.value, timestamp=sample.timestamp) + for k, mmaped_dict in six.iteritems(mmaped_dicts): + mmaped_dict.close() + dst_path = metric_type_to_dst_path[k] + # Replace existing file: + shutil.move(mmaped_dict._fname, dst_path) + + +def _is_alive(pid): + """Check to see if pid is alive""" + try: + os.kill(pid, 0) + except OSError: + return False + else: + return True + + +def archive_metrics(root=None, blocking=True, aggregate_only=False): + """Cleanup/merge database files from dead processes + + This is not threadsafe and should only be called from one thread/process at + a time (e.g. a single thread on the multiprocess exporter) + + Merges non-live metrics files from dead processes into a single file for each metric. + + In addition to merging files from dead processes, this task will collect + metrics from live metrics files, and merge them with the archived metrics, + storing the results in memory (i.e. doing the same thing as the collect() of a MultiProcessCollector. + This is value is read by the InMemoryCollector, an + alternative implementation which serves cached metrics, as opposed to + calculating them on demand, trading performance for responsiveness + + The blocking argument is mainly used for test purposes. The default + behavior is to block indefinitely, until lock acquisition. Setting + blocking=False will immediately raise an exception when acquisition fails + + The aggregate_only argument is also used for test purposes only, skipping the + merging-and-deleting process. Although it would be better to mock + _is_alive, mock is only built into python in versions 3.3 and up, and we'd + like to avoid introducing additional dependencies to this library + """ + start_time = time.time() + if root is None: + root = _multiproc_dir() + pids_to_clean = set() + live_metrics_paths = [] + + # Collect all files which belonged to dead workers + for dirname, _, filenames in os.walk(root): + for fname in filenames: + m = _db_pattern.match(fname) + if not m: + continue + name, pid = m.groups() + pid = int(pid) + pid_is_alive = _is_alive(pid) + if pid not in pids_to_clean and not pid_is_alive: + pids_to_clean.add(pid) + if pid_is_alive or aggregate_only: + live_metrics_paths.append(os.path.join(dirname, fname)) + lock_type = LOCK_EX if blocking else LOCK_EX | LOCK_NB + with advisory_lock(lock_type): + for pid in pids_to_clean: + logging.info("cleaning up worker %r", pid) + if not aggregate_only: + cleanup_process(pid) + # TODO: Skip this step if we're using a MultiprocessCollector + + # Merge metrics and cache the results + archive_paths = list(filter(os.path.exists, _get_archive_paths(root).values())) + metrics = merge(archive_paths + live_metrics_paths, accumulate=True) + time_elapsed = time.time() - start_time + _metrics_cache.write_metrics(metrics, time_elapsed) + + +@contextmanager +def advisory_lock(lock_type, filename="lockfile", prom_dir=None): + """ + Wrapper around flock. + The cleanup thread acquires an LOCK_EX + The metrics collectors acquire LOCK_SH + + The flock interface in python makes it difficult to properly time out lock + acquisition, and lock acquisition is blocking (a non-blocking lock + acquisition will immediately fail with an IOError). This should be fine, as + metrics are not exposed to the wider internet, and gets a predictable, and + low, request volume. Since they only acquire shared locks, the only + contention is with the exclusive lock acquired by the cleanup operation, + which only runs once a minute + """ + prom_dir = _multiproc_dir() if prom_dir is None else prom_dir + path = os.path.join(prom_dir, filename) + with open(path, 'w') as fd: + flock(fd, lock_type) + try: + yield + finally: + flock(fd, LOCK_UN) diff --git a/prometheus_client/multiprocess_exporter.py b/prometheus_client/multiprocess_exporter.py new file mode 100644 index 00000000..d4b39ac9 --- /dev/null +++ b/prometheus_client/multiprocess_exporter.py @@ -0,0 +1,42 @@ +import logging + +from .vendor import six + +if six.PY3: + import _thread as thread_module +else: + import thread as thread_module +import time +import traceback + +from . import (CollectorRegistry, multiprocess) +from .exposition import make_wsgi_app +from .multiprocess import archive_metrics + + +CLEANUP_INTERVAL = 5.0 + +registry = CollectorRegistry() +multiprocess.InMemoryCollector(registry) +app = make_wsgi_app(registry) +log = logging.getLogger(__name__) + + +def archive_thread(): + while True: + log.info("startup") + try: + log.info("cleaning up") + archive_metrics() + except Exception: + traceback.print_exc() + time.sleep(CLEANUP_INTERVAL) + + +def start_archiver_thread(): + thread_module.start_new_thread(archive_thread, (), {}) + + +# on_starting is a gunicorn-specific server hook +def on_starting(server): + start_archiver_thread() diff --git a/prometheus_client/multiprocess_exporter_wsgiref.py b/prometheus_client/multiprocess_exporter_wsgiref.py new file mode 100644 index 00000000..851fc2d9 --- /dev/null +++ b/prometheus_client/multiprocess_exporter_wsgiref.py @@ -0,0 +1,36 @@ +import argparse +from wsgiref.simple_server import make_server, WSGIServer + +from prometheus_client import multiprocess +from prometheus_client.exposition import make_wsgi_app +from prometheus_client.multiprocess_exporter import start_archiver_thread +from prometheus_client.registry import CollectorRegistry +""" +An entrypoint for the a multiprocess exporter using Python's built-in wsgiref implementation +The reference wsgi implementation is not pre-fork, making it more suited for the InMemoryCollector than Gunicorn +""" + +CLEANUP_INTERVAL = 5.0 + +registry = CollectorRegistry() +multiprocess.InMemoryCollector(registry) +app = make_wsgi_app(registry) + +parser = argparse.ArgumentParser(description="Starts a multiprocess prometheus exporter, running on wsgiref") +parser.add_argument("--port", type=int, required=True) +args = parser.parse_args() +port = args.port + + +class ExporterHttpServer(WSGIServer): + """ + An equivalent of the on_starting hook if running the multiprocess exporter without Gunicorn + """ + def server_activate(self): + # WSGIServer is still an old-style class in python 2.7, preventing use of super() + WSGIServer.server_activate(self) + start_archiver_thread() + + +httpd = make_server('', port, app, server_class=ExporterHttpServer) +httpd.serve_forever() diff --git a/prometheus_client/registry.py b/prometheus_client/registry.py index fa2717fb..34d6ed4f 100644 --- a/prometheus_client/registry.py +++ b/prometheus_client/registry.py @@ -2,6 +2,7 @@ from threading import Lock from .metrics_core import Metric +from .vendor import six class CollectorRegistry(object): @@ -115,6 +116,7 @@ def get_sample_value(self, name, labels=None): labels = {} for metric in self.collect(): for s in metric.samples: + assert not isinstance(s, six.text_type), s if s.name == name and s.labels == labels: return s.value return None diff --git a/prometheus_client/utils.py b/prometheus_client/utils.py index a9c9cd21..f8464cb4 100644 --- a/prometheus_client/utils.py +++ b/prometheus_client/utils.py @@ -1,5 +1,6 @@ import math + INF = float("inf") MINUS_INF = float("-inf") diff --git a/prometheus_client/values.py b/prometheus_client/values.py index 2831665a..0aba36b0 100644 --- a/prometheus_client/values.py +++ b/prometheus_client/values.py @@ -13,20 +13,27 @@ class MutexValue(object): def __init__(self, typ, metric_name, name, labelnames, labelvalues, **kwargs): self._value = 0.0 + self._timestamp = None self._lock = Lock() - def inc(self, amount): + def inc(self, amount, timestamp=None): with self._lock: self._value += amount + self._timestamp = timestamp - def set(self, value): + def set(self, value, timestamp=None): with self._lock: self._value = value + self._timestamp = timestamp def get(self): with self._lock: return self._value + def timestamp(self): + with self._lock: + return self._timestamp + def MultiProcessValue(_pidFunc=os.getpid): files = {} @@ -63,7 +70,7 @@ def __reset(self): 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) + self._value, self._timestamp = self._file.read_value_timestamp(self._key) def __check_for_pid_change(self): actual_pid = _pidFunc() @@ -76,23 +83,29 @@ def __check_for_pid_change(self): for value in values: value.__reset() - def inc(self, amount): + def inc(self, amount, timestamp=None): with lock: self.__check_for_pid_change() self._value += amount - self._file.write_value(self._key, self._value) + self._timestamp = timestamp + self._file.write_value(self._key, self._value, timestamp=self._timestamp) - def set(self, value): + def set(self, value, timestamp=None): with lock: self.__check_for_pid_change() self._value = value - self._file.write_value(self._key, self._value) + self._timestamp = timestamp + self._file.write_value(self._key, self._value, timestamp=self._timestamp) def get(self): with lock: self.__check_for_pid_change() return self._value + def timestamp(self): + with lock: + self.__check_for_pid_change() + return self._timestamp return MmapedValue diff --git a/prometheus_client/vendor/__init__.py b/prometheus_client/vendor/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/prometheus_client/vendor/six.py b/prometheus_client/vendor/six.py new file mode 100644 index 00000000..190c0239 --- /dev/null +++ b/prometheus_client/vendor/six.py @@ -0,0 +1,868 @@ +"""Utilities for writing code that runs on Python 2 and 3""" + +# Copyright (c) 2010-2015 Benjamin Peterson +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + +from __future__ import absolute_import + +import functools +import itertools +import operator +import sys +import types + +__author__ = "Benjamin Peterson " +__version__ = "1.10.0" + + +# Useful for very coarse version differentiation. +PY2 = sys.version_info[0] == 2 +PY3 = sys.version_info[0] == 3 +PY34 = sys.version_info[0:2] >= (3, 4) + +if PY3: + string_types = str, + integer_types = int, + class_types = type, + text_type = str + binary_type = bytes + + MAXSIZE = sys.maxsize +else: + string_types = basestring, + integer_types = (int, long) + class_types = (type, types.ClassType) + text_type = unicode + binary_type = str + + if sys.platform.startswith("java"): + # Jython always uses 32 bits. + MAXSIZE = int((1 << 31) - 1) + else: + # It's possible to have sizeof(long) != sizeof(Py_ssize_t). + class X(object): + + def __len__(self): + return 1 << 31 + try: + len(X()) + except OverflowError: + # 32-bit + MAXSIZE = int((1 << 31) - 1) + else: + # 64-bit + MAXSIZE = int((1 << 63) - 1) + del X + + +def _add_doc(func, doc): + """Add documentation to a function.""" + func.__doc__ = doc + + +def _import_module(name): + """Import module, returning the module after the last dot.""" + __import__(name) + return sys.modules[name] + + +class _LazyDescr(object): + + def __init__(self, name): + self.name = name + + def __get__(self, obj, tp): + result = self._resolve() + setattr(obj, self.name, result) # Invokes __set__. + try: + # This is a bit ugly, but it avoids running this again by + # removing this descriptor. + delattr(obj.__class__, self.name) + except AttributeError: + pass + return result + + +class MovedModule(_LazyDescr): + + def __init__(self, name, old, new=None): + super(MovedModule, self).__init__(name) + if PY3: + if new is None: + new = name + self.mod = new + else: + self.mod = old + + def _resolve(self): + return _import_module(self.mod) + + def __getattr__(self, attr): + _module = self._resolve() + value = getattr(_module, attr) + setattr(self, attr, value) + return value + + +class _LazyModule(types.ModuleType): + + def __init__(self, name): + super(_LazyModule, self).__init__(name) + self.__doc__ = self.__class__.__doc__ + + def __dir__(self): + attrs = ["__doc__", "__name__"] + attrs += [attr.name for attr in self._moved_attributes] + return attrs + + # Subclasses should override this + _moved_attributes = [] + + +class MovedAttribute(_LazyDescr): + + def __init__(self, name, old_mod, new_mod, old_attr=None, new_attr=None): + super(MovedAttribute, self).__init__(name) + if PY3: + if new_mod is None: + new_mod = name + self.mod = new_mod + if new_attr is None: + if old_attr is None: + new_attr = name + else: + new_attr = old_attr + self.attr = new_attr + else: + self.mod = old_mod + if old_attr is None: + old_attr = name + self.attr = old_attr + + def _resolve(self): + module = _import_module(self.mod) + return getattr(module, self.attr) + + +class _SixMetaPathImporter(object): + + """ + A meta path importer to import six.moves and its submodules. + + This class implements a PEP302 finder and loader. It should be compatible + with Python 2.5 and all existing versions of Python3 + """ + + def __init__(self, six_module_name): + self.name = six_module_name + self.known_modules = {} + + def _add_module(self, mod, *fullnames): + for fullname in fullnames: + self.known_modules[self.name + "." + fullname] = mod + + def _get_module(self, fullname): + return self.known_modules[self.name + "." + fullname] + + def find_module(self, fullname, path=None): + if fullname in self.known_modules: + return self + return None + + def __get_module(self, fullname): + try: + return self.known_modules[fullname] + except KeyError: + raise ImportError("This loader does not know module " + fullname) + + def load_module(self, fullname): + try: + # in case of a reload + return sys.modules[fullname] + except KeyError: + pass + mod = self.__get_module(fullname) + if isinstance(mod, MovedModule): + mod = mod._resolve() + else: + mod.__loader__ = self + sys.modules[fullname] = mod + return mod + + def is_package(self, fullname): + """ + Return true, if the named module is a package. + + We need this method to get correct spec objects with + Python 3.4 (see PEP451) + """ + return hasattr(self.__get_module(fullname), "__path__") + + def get_code(self, fullname): + """Return None + + Required, if is_package is implemented""" + self.__get_module(fullname) # eventually raises ImportError + return None + get_source = get_code # same as get_code + +_importer = _SixMetaPathImporter(__name__) + + +class _MovedItems(_LazyModule): + + """Lazy loading of moved objects""" + __path__ = [] # mark as package + + +_moved_attributes = [ + MovedAttribute("cStringIO", "cStringIO", "io", "StringIO"), + MovedAttribute("filter", "itertools", "builtins", "ifilter", "filter"), + MovedAttribute("filterfalse", "itertools", "itertools", "ifilterfalse", "filterfalse"), + MovedAttribute("input", "__builtin__", "builtins", "raw_input", "input"), + MovedAttribute("intern", "__builtin__", "sys"), + MovedAttribute("map", "itertools", "builtins", "imap", "map"), + MovedAttribute("getcwd", "os", "os", "getcwdu", "getcwd"), + MovedAttribute("getcwdb", "os", "os", "getcwd", "getcwdb"), + MovedAttribute("range", "__builtin__", "builtins", "xrange", "range"), + MovedAttribute("reload_module", "__builtin__", "importlib" if PY34 else "imp", "reload"), + MovedAttribute("reduce", "__builtin__", "functools"), + MovedAttribute("shlex_quote", "pipes", "shlex", "quote"), + MovedAttribute("StringIO", "StringIO", "io"), + MovedAttribute("UserDict", "UserDict", "collections"), + MovedAttribute("UserList", "UserList", "collections"), + MovedAttribute("UserString", "UserString", "collections"), + MovedAttribute("xrange", "__builtin__", "builtins", "xrange", "range"), + MovedAttribute("zip", "itertools", "builtins", "izip", "zip"), + MovedAttribute("zip_longest", "itertools", "itertools", "izip_longest", "zip_longest"), + MovedModule("builtins", "__builtin__"), + MovedModule("configparser", "ConfigParser"), + MovedModule("copyreg", "copy_reg"), + MovedModule("dbm_gnu", "gdbm", "dbm.gnu"), + MovedModule("_dummy_thread", "dummy_thread", "_dummy_thread"), + MovedModule("http_cookiejar", "cookielib", "http.cookiejar"), + MovedModule("http_cookies", "Cookie", "http.cookies"), + MovedModule("html_entities", "htmlentitydefs", "html.entities"), + MovedModule("html_parser", "HTMLParser", "html.parser"), + MovedModule("http_client", "httplib", "http.client"), + MovedModule("email_mime_multipart", "email.MIMEMultipart", "email.mime.multipart"), + MovedModule("email_mime_nonmultipart", "email.MIMENonMultipart", "email.mime.nonmultipart"), + MovedModule("email_mime_text", "email.MIMEText", "email.mime.text"), + MovedModule("email_mime_base", "email.MIMEBase", "email.mime.base"), + MovedModule("BaseHTTPServer", "BaseHTTPServer", "http.server"), + MovedModule("CGIHTTPServer", "CGIHTTPServer", "http.server"), + MovedModule("SimpleHTTPServer", "SimpleHTTPServer", "http.server"), + MovedModule("cPickle", "cPickle", "pickle"), + MovedModule("queue", "Queue"), + MovedModule("reprlib", "repr"), + MovedModule("socketserver", "SocketServer"), + MovedModule("_thread", "thread", "_thread"), + MovedModule("tkinter", "Tkinter"), + MovedModule("tkinter_dialog", "Dialog", "tkinter.dialog"), + MovedModule("tkinter_filedialog", "FileDialog", "tkinter.filedialog"), + MovedModule("tkinter_scrolledtext", "ScrolledText", "tkinter.scrolledtext"), + MovedModule("tkinter_simpledialog", "SimpleDialog", "tkinter.simpledialog"), + MovedModule("tkinter_tix", "Tix", "tkinter.tix"), + MovedModule("tkinter_ttk", "ttk", "tkinter.ttk"), + MovedModule("tkinter_constants", "Tkconstants", "tkinter.constants"), + MovedModule("tkinter_dnd", "Tkdnd", "tkinter.dnd"), + MovedModule("tkinter_colorchooser", "tkColorChooser", + "tkinter.colorchooser"), + MovedModule("tkinter_commondialog", "tkCommonDialog", + "tkinter.commondialog"), + MovedModule("tkinter_tkfiledialog", "tkFileDialog", "tkinter.filedialog"), + MovedModule("tkinter_font", "tkFont", "tkinter.font"), + MovedModule("tkinter_messagebox", "tkMessageBox", "tkinter.messagebox"), + MovedModule("tkinter_tksimpledialog", "tkSimpleDialog", + "tkinter.simpledialog"), + MovedModule("urllib_parse", __name__ + ".moves.urllib_parse", "urllib.parse"), + MovedModule("urllib_error", __name__ + ".moves.urllib_error", "urllib.error"), + MovedModule("urllib", __name__ + ".moves.urllib", __name__ + ".moves.urllib"), + MovedModule("urllib_robotparser", "robotparser", "urllib.robotparser"), + MovedModule("xmlrpc_client", "xmlrpclib", "xmlrpc.client"), + MovedModule("xmlrpc_server", "SimpleXMLRPCServer", "xmlrpc.server"), +] +# Add windows specific modules. +if sys.platform == "win32": + _moved_attributes += [ + MovedModule("winreg", "_winreg"), + ] + +for attr in _moved_attributes: + setattr(_MovedItems, attr.name, attr) + if isinstance(attr, MovedModule): + _importer._add_module(attr, "moves." + attr.name) +del attr + +_MovedItems._moved_attributes = _moved_attributes + +moves = _MovedItems(__name__ + ".moves") +_importer._add_module(moves, "moves") + + +class Module_six_moves_urllib_parse(_LazyModule): + + """Lazy loading of moved objects in six.moves.urllib_parse""" + + +_urllib_parse_moved_attributes = [ + MovedAttribute("ParseResult", "urlparse", "urllib.parse"), + MovedAttribute("SplitResult", "urlparse", "urllib.parse"), + MovedAttribute("parse_qs", "urlparse", "urllib.parse"), + MovedAttribute("parse_qsl", "urlparse", "urllib.parse"), + MovedAttribute("urldefrag", "urlparse", "urllib.parse"), + MovedAttribute("urljoin", "urlparse", "urllib.parse"), + MovedAttribute("urlparse", "urlparse", "urllib.parse"), + MovedAttribute("urlsplit", "urlparse", "urllib.parse"), + MovedAttribute("urlunparse", "urlparse", "urllib.parse"), + MovedAttribute("urlunsplit", "urlparse", "urllib.parse"), + MovedAttribute("quote", "urllib", "urllib.parse"), + MovedAttribute("quote_plus", "urllib", "urllib.parse"), + MovedAttribute("unquote", "urllib", "urllib.parse"), + MovedAttribute("unquote_plus", "urllib", "urllib.parse"), + MovedAttribute("urlencode", "urllib", "urllib.parse"), + MovedAttribute("splitquery", "urllib", "urllib.parse"), + MovedAttribute("splittag", "urllib", "urllib.parse"), + MovedAttribute("splituser", "urllib", "urllib.parse"), + MovedAttribute("uses_fragment", "urlparse", "urllib.parse"), + MovedAttribute("uses_netloc", "urlparse", "urllib.parse"), + MovedAttribute("uses_params", "urlparse", "urllib.parse"), + MovedAttribute("uses_query", "urlparse", "urllib.parse"), + MovedAttribute("uses_relative", "urlparse", "urllib.parse"), +] +for attr in _urllib_parse_moved_attributes: + setattr(Module_six_moves_urllib_parse, attr.name, attr) +del attr + +Module_six_moves_urllib_parse._moved_attributes = _urllib_parse_moved_attributes + +_importer._add_module(Module_six_moves_urllib_parse(__name__ + ".moves.urllib_parse"), + "moves.urllib_parse", "moves.urllib.parse") + + +class Module_six_moves_urllib_error(_LazyModule): + + """Lazy loading of moved objects in six.moves.urllib_error""" + + +_urllib_error_moved_attributes = [ + MovedAttribute("URLError", "urllib2", "urllib.error"), + MovedAttribute("HTTPError", "urllib2", "urllib.error"), + MovedAttribute("ContentTooShortError", "urllib", "urllib.error"), +] +for attr in _urllib_error_moved_attributes: + setattr(Module_six_moves_urllib_error, attr.name, attr) +del attr + +Module_six_moves_urllib_error._moved_attributes = _urllib_error_moved_attributes + +_importer._add_module(Module_six_moves_urllib_error(__name__ + ".moves.urllib.error"), + "moves.urllib_error", "moves.urllib.error") + + +class Module_six_moves_urllib_request(_LazyModule): + + """Lazy loading of moved objects in six.moves.urllib_request""" + + +_urllib_request_moved_attributes = [ + MovedAttribute("urlopen", "urllib2", "urllib.request"), + MovedAttribute("install_opener", "urllib2", "urllib.request"), + MovedAttribute("build_opener", "urllib2", "urllib.request"), + MovedAttribute("pathname2url", "urllib", "urllib.request"), + MovedAttribute("url2pathname", "urllib", "urllib.request"), + MovedAttribute("getproxies", "urllib", "urllib.request"), + MovedAttribute("Request", "urllib2", "urllib.request"), + MovedAttribute("OpenerDirector", "urllib2", "urllib.request"), + MovedAttribute("HTTPDefaultErrorHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPRedirectHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPCookieProcessor", "urllib2", "urllib.request"), + MovedAttribute("ProxyHandler", "urllib2", "urllib.request"), + MovedAttribute("BaseHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPPasswordMgr", "urllib2", "urllib.request"), + MovedAttribute("HTTPPasswordMgrWithDefaultRealm", "urllib2", "urllib.request"), + MovedAttribute("AbstractBasicAuthHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPBasicAuthHandler", "urllib2", "urllib.request"), + MovedAttribute("ProxyBasicAuthHandler", "urllib2", "urllib.request"), + MovedAttribute("AbstractDigestAuthHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPDigestAuthHandler", "urllib2", "urllib.request"), + MovedAttribute("ProxyDigestAuthHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPSHandler", "urllib2", "urllib.request"), + MovedAttribute("FileHandler", "urllib2", "urllib.request"), + MovedAttribute("FTPHandler", "urllib2", "urllib.request"), + MovedAttribute("CacheFTPHandler", "urllib2", "urllib.request"), + MovedAttribute("UnknownHandler", "urllib2", "urllib.request"), + MovedAttribute("HTTPErrorProcessor", "urllib2", "urllib.request"), + MovedAttribute("urlretrieve", "urllib", "urllib.request"), + MovedAttribute("urlcleanup", "urllib", "urllib.request"), + MovedAttribute("URLopener", "urllib", "urllib.request"), + MovedAttribute("FancyURLopener", "urllib", "urllib.request"), + MovedAttribute("proxy_bypass", "urllib", "urllib.request"), +] +for attr in _urllib_request_moved_attributes: + setattr(Module_six_moves_urllib_request, attr.name, attr) +del attr + +Module_six_moves_urllib_request._moved_attributes = _urllib_request_moved_attributes + +_importer._add_module(Module_six_moves_urllib_request(__name__ + ".moves.urllib.request"), + "moves.urllib_request", "moves.urllib.request") + + +class Module_six_moves_urllib_response(_LazyModule): + + """Lazy loading of moved objects in six.moves.urllib_response""" + + +_urllib_response_moved_attributes = [ + MovedAttribute("addbase", "urllib", "urllib.response"), + MovedAttribute("addclosehook", "urllib", "urllib.response"), + MovedAttribute("addinfo", "urllib", "urllib.response"), + MovedAttribute("addinfourl", "urllib", "urllib.response"), +] +for attr in _urllib_response_moved_attributes: + setattr(Module_six_moves_urllib_response, attr.name, attr) +del attr + +Module_six_moves_urllib_response._moved_attributes = _urllib_response_moved_attributes + +_importer._add_module(Module_six_moves_urllib_response(__name__ + ".moves.urllib.response"), + "moves.urllib_response", "moves.urllib.response") + + +class Module_six_moves_urllib_robotparser(_LazyModule): + + """Lazy loading of moved objects in six.moves.urllib_robotparser""" + + +_urllib_robotparser_moved_attributes = [ + MovedAttribute("RobotFileParser", "robotparser", "urllib.robotparser"), +] +for attr in _urllib_robotparser_moved_attributes: + setattr(Module_six_moves_urllib_robotparser, attr.name, attr) +del attr + +Module_six_moves_urllib_robotparser._moved_attributes = _urllib_robotparser_moved_attributes + +_importer._add_module(Module_six_moves_urllib_robotparser(__name__ + ".moves.urllib.robotparser"), + "moves.urllib_robotparser", "moves.urllib.robotparser") + + +class Module_six_moves_urllib(types.ModuleType): + + """Create a six.moves.urllib namespace that resembles the Python 3 namespace""" + __path__ = [] # mark as package + parse = _importer._get_module("moves.urllib_parse") + error = _importer._get_module("moves.urllib_error") + request = _importer._get_module("moves.urllib_request") + response = _importer._get_module("moves.urllib_response") + robotparser = _importer._get_module("moves.urllib_robotparser") + + def __dir__(self): + return ['parse', 'error', 'request', 'response', 'robotparser'] + +_importer._add_module(Module_six_moves_urllib(__name__ + ".moves.urllib"), + "moves.urllib") + + +def add_move(move): + """Add an item to six.moves.""" + setattr(_MovedItems, move.name, move) + + +def remove_move(name): + """Remove item from six.moves.""" + try: + delattr(_MovedItems, name) + except AttributeError: + try: + del moves.__dict__[name] + except KeyError: + raise AttributeError("no such move, %r" % (name,)) + + +if PY3: + _meth_func = "__func__" + _meth_self = "__self__" + + _func_closure = "__closure__" + _func_code = "__code__" + _func_defaults = "__defaults__" + _func_globals = "__globals__" +else: + _meth_func = "im_func" + _meth_self = "im_self" + + _func_closure = "func_closure" + _func_code = "func_code" + _func_defaults = "func_defaults" + _func_globals = "func_globals" + + +try: + advance_iterator = next +except NameError: + def advance_iterator(it): + return it.next() +next = advance_iterator + + +try: + callable = callable +except NameError: + def callable(obj): + return any("__call__" in klass.__dict__ for klass in type(obj).__mro__) + + +if PY3: + def get_unbound_function(unbound): + return unbound + + create_bound_method = types.MethodType + + def create_unbound_method(func, cls): + return func + + Iterator = object +else: + def get_unbound_function(unbound): + return unbound.im_func + + def create_bound_method(func, obj): + return types.MethodType(func, obj, obj.__class__) + + def create_unbound_method(func, cls): + return types.MethodType(func, None, cls) + + class Iterator(object): + + def next(self): + return type(self).__next__(self) + + callable = callable +_add_doc(get_unbound_function, + """Get the function out of a possibly unbound function""") + + +get_method_function = operator.attrgetter(_meth_func) +get_method_self = operator.attrgetter(_meth_self) +get_function_closure = operator.attrgetter(_func_closure) +get_function_code = operator.attrgetter(_func_code) +get_function_defaults = operator.attrgetter(_func_defaults) +get_function_globals = operator.attrgetter(_func_globals) + + +if PY3: + def iterkeys(d, **kw): + return iter(d.keys(**kw)) + + def itervalues(d, **kw): + return iter(d.values(**kw)) + + def iteritems(d, **kw): + return iter(d.items(**kw)) + + def iterlists(d, **kw): + return iter(d.lists(**kw)) + + viewkeys = operator.methodcaller("keys") + + viewvalues = operator.methodcaller("values") + + viewitems = operator.methodcaller("items") +else: + def iterkeys(d, **kw): + return d.iterkeys(**kw) + + def itervalues(d, **kw): + return d.itervalues(**kw) + + def iteritems(d, **kw): + return d.iteritems(**kw) + + def iterlists(d, **kw): + return d.iterlists(**kw) + + viewkeys = operator.methodcaller("viewkeys") + + viewvalues = operator.methodcaller("viewvalues") + + viewitems = operator.methodcaller("viewitems") + +_add_doc(iterkeys, "Return an iterator over the keys of a dictionary.") +_add_doc(itervalues, "Return an iterator over the values of a dictionary.") +_add_doc(iteritems, + "Return an iterator over the (key, value) pairs of a dictionary.") +_add_doc(iterlists, + "Return an iterator over the (key, [values]) pairs of a dictionary.") + + +if PY3: + def b(s): + return s.encode("latin-1") + + def u(s): + return s + unichr = chr + import struct + int2byte = struct.Struct(">B").pack + del struct + byte2int = operator.itemgetter(0) + indexbytes = operator.getitem + iterbytes = iter + import io + StringIO = io.StringIO + BytesIO = io.BytesIO + _assertCountEqual = "assertCountEqual" + if sys.version_info[1] <= 1: + _assertRaisesRegex = "assertRaisesRegexp" + _assertRegex = "assertRegexpMatches" + else: + _assertRaisesRegex = "assertRaisesRegex" + _assertRegex = "assertRegex" +else: + def b(s): + return s + # Workaround for standalone backslash + + def u(s): + return unicode(s.replace(r'\\', r'\\\\'), "unicode_escape") + unichr = unichr + int2byte = chr + + def byte2int(bs): + return ord(bs[0]) + + def indexbytes(buf, i): + return ord(buf[i]) + iterbytes = functools.partial(itertools.imap, ord) + import StringIO + StringIO = BytesIO = StringIO.StringIO + _assertCountEqual = "assertItemsEqual" + _assertRaisesRegex = "assertRaisesRegexp" + _assertRegex = "assertRegexpMatches" +_add_doc(b, """Byte literal""") +_add_doc(u, """Text literal""") + + +def assertCountEqual(self, *args, **kwargs): + return getattr(self, _assertCountEqual)(*args, **kwargs) + + +def assertRaisesRegex(self, *args, **kwargs): + return getattr(self, _assertRaisesRegex)(*args, **kwargs) + + +def assertRegex(self, *args, **kwargs): + return getattr(self, _assertRegex)(*args, **kwargs) + + +if PY3: + exec_ = getattr(moves.builtins, "exec") + + def reraise(tp, value, tb=None): + if value is None: + value = tp() + if value.__traceback__ is not tb: + raise value.with_traceback(tb) + raise value + +else: + def exec_(_code_, _globs_=None, _locs_=None): + """Execute code in a namespace.""" + if _globs_ is None: + frame = sys._getframe(1) + _globs_ = frame.f_globals + if _locs_ is None: + _locs_ = frame.f_locals + del frame + elif _locs_ is None: + _locs_ = _globs_ + exec("""exec _code_ in _globs_, _locs_""") + + exec_("""def reraise(tp, value, tb=None): + raise tp, value, tb +""") + + +if sys.version_info[:2] == (3, 2): + exec_("""def raise_from(value, from_value): + if from_value is None: + raise value + raise value from from_value +""") +elif sys.version_info[:2] > (3, 2): + exec_("""def raise_from(value, from_value): + raise value from from_value +""") +else: + def raise_from(value, from_value): + raise value + + +print_ = getattr(moves.builtins, "print", None) +if print_ is None: + def print_(*args, **kwargs): + """The new-style print function for Python 2.4 and 2.5.""" + fp = kwargs.pop("file", sys.stdout) + if fp is None: + return + + def write(data): + if not isinstance(data, basestring): + data = str(data) + # If the file has an encoding, encode unicode with it. + if (isinstance(fp, file) and + isinstance(data, unicode) and + fp.encoding is not None): + errors = getattr(fp, "errors", None) + if errors is None: + errors = "strict" + data = data.encode(fp.encoding, errors) + fp.write(data) + want_unicode = False + sep = kwargs.pop("sep", None) + if sep is not None: + if isinstance(sep, unicode): + want_unicode = True + elif not isinstance(sep, str): + raise TypeError("sep must be None or a string") + end = kwargs.pop("end", None) + if end is not None: + if isinstance(end, unicode): + want_unicode = True + elif not isinstance(end, str): + raise TypeError("end must be None or a string") + if kwargs: + raise TypeError("invalid keyword arguments to print()") + if not want_unicode: + for arg in args: + if isinstance(arg, unicode): + want_unicode = True + break + if want_unicode: + newline = unicode("\n") + space = unicode(" ") + else: + newline = "\n" + space = " " + if sep is None: + sep = space + if end is None: + end = newline + for i, arg in enumerate(args): + if i: + write(sep) + write(arg) + write(end) +if sys.version_info[:2] < (3, 3): + _print = print_ + + def print_(*args, **kwargs): + fp = kwargs.get("file", sys.stdout) + flush = kwargs.pop("flush", False) + _print(*args, **kwargs) + if flush and fp is not None: + fp.flush() + +_add_doc(reraise, """Reraise an exception.""") + +if sys.version_info[0:2] < (3, 4): + def wraps(wrapped, assigned=functools.WRAPPER_ASSIGNMENTS, + updated=functools.WRAPPER_UPDATES): + def wrapper(f): + f = functools.wraps(wrapped, assigned, updated)(f) + f.__wrapped__ = wrapped + return f + return wrapper +else: + wraps = functools.wraps + + +def with_metaclass(meta, *bases): + """Create a base class with a metaclass.""" + # This requires a bit of explanation: the basic idea is to make a dummy + # metaclass for one level of class instantiation that replaces itself with + # the actual metaclass. + class metaclass(meta): + + def __new__(cls, name, this_bases, d): + return meta(name, bases, d) + return type.__new__(metaclass, 'temporary_class', (), {}) + + +def add_metaclass(metaclass): + """Class decorator for creating a class with a metaclass.""" + def wrapper(cls): + orig_vars = cls.__dict__.copy() + slots = orig_vars.get('__slots__') + if slots is not None: + if isinstance(slots, str): + slots = [slots] + for slots_var in slots: + orig_vars.pop(slots_var) + orig_vars.pop('__dict__', None) + orig_vars.pop('__weakref__', None) + return metaclass(cls.__name__, cls.__bases__, orig_vars) + return wrapper + + +def python_2_unicode_compatible(klass): + """ + A decorator that defines __unicode__ and __str__ methods under Python 2. + Under Python 3 it does nothing. + + To support Python 2 and 3 with a single code base, define a __str__ method + returning text and apply this decorator to the class. + """ + if PY2: + if '__str__' not in klass.__dict__: + raise ValueError("@python_2_unicode_compatible cannot be applied " + "to %s because it doesn't define __str__()." % + klass.__name__) + klass.__unicode__ = klass.__str__ + klass.__str__ = lambda self: self.__unicode__().encode('utf-8') + return klass + + +# Complete the moves implementation. +# This code is at the end of this module to speed up module loading. +# Turn this module into a package. +__path__ = [] # required for PEP 302 and PEP 451 +__package__ = __name__ # see PEP 366 @ReservedAssignment +if globals().get("__spec__") is not None: + __spec__.submodule_search_locations = [] # PEP 451 @UndefinedVariable +# Remove other six meta path importers, since they cause problems. This can +# happen if six is removed from sys.modules and then reloaded. (Setuptools does +# this for some reason.) +if sys.meta_path: + for i, importer in enumerate(sys.meta_path): + # Here's some real nastiness: Another "instance" of the six module might + # be floating around. Therefore, we can't use isinstance() to check for + # the six meta path importer, since the other six instance will have + # inserted an importer with different class. + if (type(importer).__name__ == "_SixMetaPathImporter" and + importer.name == __name__): + del sys.meta_path[i] + break + del i, importer +# Finally, add the importer to the meta path import hook. +sys.meta_path.append(_importer) diff --git a/setup.py b/setup.py index 73b5048d..7efdfb7b 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ setup( name="prometheus_client", - version="0.6.0", + version="0.8.4", author="Brian Brazil", author_email="brian.brazil@robustperception.io", description="Python client for the Prometheus monitoring system.", @@ -17,6 +17,7 @@ 'prometheus_client.bridge', 'prometheus_client.openmetrics', 'prometheus_client.twisted', + 'prometheus_client.vendor' ], extras_require={ 'twisted': ['twisted'], diff --git a/tests/openmetrics/test_exposition.py b/tests/openmetrics/test_exposition.py index 502a45e0..5e8622f2 100644 --- a/tests/openmetrics/test_exposition.py +++ b/tests/openmetrics/test_exposition.py @@ -49,7 +49,7 @@ def test_counter_total(self): generate_latest(self.registry)) def test_gauge(self): - g = Gauge('gg', 'A gauge', registry=self.registry) + g = Gauge('gg', 'A gauge', registry=self.registry, multiprocess_mode='all') g.set(17) self.assertEqual(b'# HELP gg A gauge\n# TYPE gg gauge\ngg 17.0\n# EOF\n', generate_latest(self.registry)) diff --git a/tests/test_exposition.py b/tests/test_exposition.py index 00f39b47..89858043 100644 --- a/tests/test_exposition.py +++ b/tests/test_exposition.py @@ -71,7 +71,8 @@ def test_counter_total(self): """, generate_latest(self.registry)) def test_gauge(self): - g = Gauge('gg', 'A gauge', registry=self.registry) + g = Gauge('gg', 'A gauge', registry=self.registry, + multiprocess_mode='all') g.set(17) self.assertEqual(b'# HELP gg A gauge\n# TYPE gg gauge\ngg 17.0\n', generate_latest(self.registry)) @@ -139,13 +140,15 @@ def test_enum(self): generate_latest(self.registry)) def test_unicode(self): - c = Gauge('cc', '\u4500', ['l'], registry=self.registry) + c = Gauge('cc', '\u4500', ['l'], registry=self.registry, + multiprocess_mode='all') c.labels('\u4500').inc() self.assertEqual(b'# HELP cc \xe4\x94\x80\n# TYPE cc gauge\ncc{l="\xe4\x94\x80"} 1.0\n', generate_latest(self.registry)) def test_escaping(self): - g = Gauge('cc', 'A\ngaug\\e', ['a'], registry=self.registry) + g = Gauge('cc', 'A\ngaug\\e', ['a'], registry=self.registry, + multiprocess_mode='all') g.labels('\\x\n"').inc(1) self.assertEqual(b'# HELP cc A\\ngaug\\\\e\n# TYPE cc gauge\ncc{a="\\\\x\\n\\""} 1.0\n', generate_latest(self.registry)) diff --git a/tests/test_gc_collector.py b/tests/test_gc_collector.py index fd10f277..c0cda88f 100644 --- a/tests/test_gc_collector.py +++ b/tests/test_gc_collector.py @@ -1,8 +1,8 @@ from __future__ import unicode_literals import gc -import sys import platform +import sys if sys.version_info < (2, 7): # We need the skip decorators from unittest2 on Python 2.6. diff --git a/tests/test_multiprocess.py b/tests/test_multiprocess.py index be031524..ffdcc0b7 100644 --- a/tests/test_multiprocess.py +++ b/tests/test_multiprocess.py @@ -1,17 +1,22 @@ from __future__ import unicode_literals +from fcntl import LOCK_EX, LOCK_SH import glob import os import shutil import sys import tempfile +import time from prometheus_client import mmap_dict, values from prometheus_client.core import ( CollectorRegistry, Counter, Gauge, Histogram, Sample, Summary, ) +from prometheus_client.exposition import generate_latest +import prometheus_client.multiprocess from prometheus_client.multiprocess import ( - mark_process_dead, MultiProcessCollector, + advisory_lock, archive_metrics, InMemoryCollector, mark_process_dead, + merge, MultiProcessCollector ) from prometheus_client.values import MultiProcessValue, MutexValue @@ -73,31 +78,55 @@ def test_histogram_adds(self): 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 = MultiProcessValue(lambda: 123) + g1 = Gauge('g', 'help', registry=None, multiprocess_mode='all') values.ValueClass = MultiProcessValue(lambda: 456) - g2 = Gauge('g', 'help', registry=None) + g2 = Gauge('g', 'help', registry=None, multiprocess_mode='all') 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) + archive_metrics() 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') + self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'})) + g1.set(1) values.ValueClass = MultiProcessValue(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_latest(self): + self.assertEqual(None, self.registry.get_sample_value('g')) + g1 = Gauge('g', 'G', registry=None, multiprocess_mode=Gauge.LATEST) + g1.set(0) + self.assertEqual(0, self.registry.get_sample_value('g')) + g1.set(123) + self.assertEqual(123, self.registry.get_sample_value('g')) + + t0 = time.time() + g1.set(1, timestamp=t0) + self.assertEqual(1, self.registry.get_sample_value('g')) + archive_metrics() + self.assertEqual(1, self.registry.get_sample_value('g')) + values.ValueClass = MultiProcessValue(lambda: '456789') + g2 = Gauge('g', 'G', registry=None, multiprocess_mode=Gauge.LATEST) + t1 = t0 - time.time() + g2.set(2, timestamp=t1) + self.assertEqual(1, self.registry.get_sample_value('g')) + archive_metrics() + self.assertEqual(1, self.registry.get_sample_value('g')) + def test_gauge_min(self): g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min') values.ValueClass = MultiProcessValue(lambda: 456) @@ -175,7 +204,8 @@ def add_label(key, value): return l c = Counter('c', 'help', labelnames=labels.keys(), registry=None) - g = Gauge('g', 'help', labelnames=labels.keys(), registry=None) + g = Gauge('g', 'help', labelnames=labels.keys(), registry=None, + multiprocess_mode='all') h = Histogram('h', 'help', labelnames=labels.keys(), registry=None) c.labels(**labels).inc(1) @@ -243,7 +273,7 @@ def add_label(key, value): 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) + (m.name, m) for m in merge(files, accumulate=False) ) metrics['h'].samples.sort( @@ -271,23 +301,41 @@ def add_label(key, value): self.assertEqual(metrics['h'].samples, expected_histogram) + def test_missing_gauge_file_during_merge(self): + # These files don't exist, just like if mark_process_dead(9999999) had been + # called during self.collector.collect(), after the glob found it + # but before the merge actually happened. + # This should not raise and return no metrics + self.assertFalse(merge([ + os.path.join(self.tempdir, 'gauge_liveall_9999999.db'), + os.path.join(self.tempdir, 'gauge_livesum_9999999.db'), + ])) + + class TestMmapedDict(unittest.TestCase): def setUp(self): fd, self.tempfile = tempfile.mkstemp() os.close(fd) self.d = mmap_dict.MmapedDict(self.tempfile) + def test_timestamp(self): + t0 = time.time() + self.d.write_value("foo", 3.0, timestamp=t0) + v, t = self.d.read_value_timestamp("foo") + self.assertTrue((v - 3.0) ** 2 < 0.001) + self.assertTrue((t0 - t) ** 2 < 0.001) + 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())) + self.assertEqual([('abc', 123.0, None)], 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())) + self.assertEqual([(key, 123.0, None)], list(self.d.read_all_values())) def test_multi_expansion(self): key = 'a' * mmap_dict._INITIAL_MMAP_SIZE * 4 @@ -295,7 +343,7 @@ def test_multi_expansion(self): 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)], + [('abc', 42.0, None), (key, 123.0, None), ('def', 17.0, None)], list(self.d.read_all_values())) def test_corruption_detected(self): @@ -326,3 +374,135 @@ def test_file_syncpath(self): def tearDown(self): os.remove(self.tmpfl) + + +class TestAdvisoryLock(unittest.TestCase): + """ + These tests use lock aqusition as a proxy for cleanup/collect operations, + the former using exclusive locks, the latter shared locks + """ + def setUp(self): + self.tempdir = tempfile.mkdtemp() + os.environ['prometheus_multiproc_dir'] = self.tempdir + values.ValueClass = MultiProcessValue(lambda: 123) + self.registry = CollectorRegistry() + self.collector = MultiProcessCollector(self.registry, self.tempdir) + + def test_cleanup_waits_for_collectors(self): + # IOError in python2, OSError in python3 + with self.assertRaises(EnvironmentError): + with advisory_lock(LOCK_SH): + archive_metrics(blocking=False) + + def test_collect_doesnt_block_other_collects(self): + values.ValueClass = MultiProcessValue(lambda: 0) + labels = dict((i, i) for i in 'abcd') + c = Counter('c', 'help', labelnames=labels.keys(), registry=None) + c.labels(**labels).inc(1) + + with advisory_lock(LOCK_SH): + metrics = dict((m.name, m) for m in self.collector.collect(blocking=False)) + self.assertEqual( + metrics['c'].samples, [Sample('c_total', labels, 1.0)] + ) + + def test_collect_waits_for_cleanup(self): + values.ValueClass = MultiProcessValue(lambda: 0) + labels = dict((i, i) for i in 'abcd') + c = Counter('c', 'help', labelnames=labels.keys(), registry=None) + c.labels(**labels).inc(1) + with self.assertRaises(EnvironmentError): + with advisory_lock(LOCK_EX): + self.collector.collect(blocking=False) + + def test_exceptions_release_lock(self): + with self.assertRaises(ValueError): + with advisory_lock(LOCK_EX): + raise ValueError + # Do an operation which requires acquiring the lock + archive_metrics(blocking=False) + + def tearDown(self): + del os.environ['prometheus_multiproc_dir'] + shutil.rmtree(self.tempdir) + values.ValueClass = MutexValue + + +class TestInMemoryCollector(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.mkdtemp() + os.environ['prometheus_multiproc_dir'] = self.tempdir + values.ValueClass = MultiProcessValue(lambda: 123) + self.registry = CollectorRegistry() + self.collector = InMemoryCollector(self.registry) + + def tearDown(self): + del os.environ['prometheus_multiproc_dir'] + shutil.rmtree(self.tempdir) + values.ValueClass = MutexValue + prometheus_client.multiprocess._metrics_cache = prometheus_client.multiprocess.MetricsCache() + + def test_serves_empty_metrics_if_no_metrics_written(self): + self.assertEqual(self.collector.collect(), []) + + def test_serves_empty_metrics_if_not_processed(self): + c1 = Counter('c', 'help', registry=None) + # The cleanup/archiver task hasn't run yet, no metrics + self.assertEqual(None, self.registry.get_sample_value('c_total')) + c1.inc(1) + # Still no metrics + self.assertEqual(self.collector.collect(), []) + + def test_serves_metrics(self): + labels = dict((i, i) for i in 'abcd') + c = Counter('c', 'help', labelnames=labels.keys(), registry=None) + c.labels(**labels).inc(1) + self.assertEqual(None, self.registry.get_sample_value('c_total', labels)) + archive_metrics() + self.assertEqual(self.collector.collect()[0].samples, [Sample('c_total', labels, 1.0)]) + + def test_displays_archive_stats(self): + output = generate_latest(self.registry) + self.assertIn("archive_duration_seconds", output) + + def test_aggregates_live_and_archived_metrics(self): + pid = 456 + values.ValueClass = MultiProcessValue(lambda: pid) + + def files(): + fs = os.listdir(os.environ['prometheus_multiproc_dir']) + fs.sort() + return fs + c1 = Counter('c1', 'c1', registry=None) + c1.inc(1) + self.assertIn('counter_456.db', files()) + + archive_metrics() + self.assertNotIn('counter_456.db', files()) + self.assertEqual(1, self.registry.get_sample_value('c1_total')) + + pid = 789 + values.ValueClass = MultiProcessValue(lambda: pid) + c1 = Counter('c1', 'c1', registry=None) + c1.inc(2) + g1 = Gauge('g1', 'g1', registry=None, multiprocess_mode="liveall") + g1.set(5) + self.assertIn('counter_789.db', files()) + # Pretend that pid 789 is live + archive_metrics(aggregate_only=True) + + # The live counter should be merged with the archived counter, and the + # liveall gauge should be included + self.assertIn('counter_789.db', files()) + self.assertIn('gauge_liveall_789.db', files()) + self.assertEqual(3, self.registry.get_sample_value('c1_total')) + self.assertEqual(5, self.registry.get_sample_value('g1', labels={u'pid': u'789'})) + # Now pid 789 is dead + archive_metrics() + + # The formerly live counter's value should be archived, and the + # liveall gauge should be removed completely + self.assertNotIn('counter_789.db', files()) + self.assertNotIn('gauge_liveall_789.db', files()) + self.assertEqual(3, self.registry.get_sample_value('c1_total')) + self.assertEqual(None, self.registry.get_sample_value('g1', labels={u'pid': u'789'})) diff --git a/tox.ini b/tox.ini index 8f7f81b3..440f655b 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,6 @@ [tox] -envlist = coverage-clean,py26,py27,py34,py35,py36,pypy,pypy3,{py27,py36}-nooptionals,coverage-report,flake8 +# envlist = coverage-clean,py26,py27,py34,py35,py36,pypy,pypy3,{py27,py36}-nooptionals,coverage-report,flake8 +envlist = coverage-clean,py27,coverage-report,flake8 [base]