Add multi-process support.

This works by having a shelve per process that are continously updated
with metrics, and a collector that reads from them.

Histogram buckets need accumulation and special handling for _count,
and for gauges we need to offer a number of options.

Author: brian-brazil

README.md

@@ -365,6 +365,74 @@ REGISTRY.register(CustomCollector())
 
 `SummaryMetricFamily` and `HistogramMetricFamily` work similarly.
 
+## Multiprocess Mode (Gunicorn)
+
+**Experimental: This feature is new and has rough edges.**
+
+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
+it's common to have processes rather than threads to handle large workloads.
+
+To handle this the client library can be put in multiprocess mode.
+This comes with a number of limitations:
+
+- Registries can not be used as normal, all instantiated metrics are exported
+- Custom collectors do not work (e.g. cpu and memory metrics)
+- The pushgateway cannot be used
+- Gauges cannot use the `pid` label
+- Gunicron's `preload_app` feature is not supported
+
+There's several steps to getting this working:
+
+**One**: Gunicorn 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).
+
+Put the following in the config file:
+```python
+def worker_exit(server, worker):
+    from prometheus_client import multiprocess
+    multiprocess.mark_process_dead(worker.pid)
+```
+
+**Two**: Inside the application
+```python
+from prometheus_client import multiprocess
+from prometheus_client import generate_latest, CollectorRegistry, CONTENT_TYPE_LATEST, Gauge
+
+# Example gauge.
+IN_PROGRESS = Gauge("inprogress_requests", "help", multiprocess_mode='livesum')
+
+
+# Expose metrics.
+@IN_PROGRESS.track_inprogress()
+def app(environ, start_response):
+    registry = CollectorRegistry()
+    multiprocess.MultiProcessCollector(registry)
+    data = generate_latest(registry)
+    status = '200 OK'
+    response_headers = [
+        ('Content-type', CONTENT_TYPE_LATEST),
+        ('Content-Length', str(len(data)))
+    ]
+    start_response(status, response_headers)
+    return iter([data])
+```
+
+**Three**: Instrumentation
+
+Counters, Summarys and Histograms work as normal.
+
+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.
+- '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.
+- 'min': Return a single timeseries that is the minimum of the values of all processes, alive or dead.
 
 ## Parser
 

prometheus_client/core.py

@@ -4,7 +4,10 @@
 
 import copy
 import math
+import json
+import os
 import re
+import shelve
 import types
 from timeit import default_timer
 
@@ -220,7 +223,9 @@ def add_metric(self, labels, buckets, sum_value):
 class _MutexValue(object):
     '''A float protected by a mutex.'''
 
-    def __init__(self, name, labelnames, labelvalues):
+    _multiprocess = False
+
+    def __init__(self, typ, metric_name, name, labelnames, labelvalues, **kwargs):
       self._value = 0.0
       self._lock = Lock()
 
@@ -236,7 +241,60 @@ def get(self):
       with self._lock:
           return self._value
 
-_ValueClass = _MutexValue
+
+def _MultiProcessValue(__pid=os.getpid()):
+    pid = __pid
+    samples = {}
+    samples_lock = Lock()
+
+    class _ShelveValue(object):
+        '''A float protected by a mutex backed by a per-process shelve.'''
+
+        _multiprocess = True
+
+        def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', **kwargs):
+            with samples_lock:
+                if typ == 'gauge':
+                    file_prefix = typ + '_' +  multiprocess_mode
+                else:
+                    file_prefix = typ
+                if file_prefix not in samples:
+                    filename = os.path.join(os.environ['prometheus_multiproc_dir'], '{0}_{1}.db'.format(file_prefix, pid))
+                    samples[file_prefix] = shelve.open(filename)
+            self._samples = samples[file_prefix]
+            self._key = json.dumps((metric_name, name, labelnames, labelvalues))
+            self._value = self._samples.get(self._key, 0.0)
+            self._samples[self._key] = self._value
+            self._samples.sync()
+            self._lock = Lock()
+
+        def inc(self, amount):
+            with self._lock:
+                self._value += amount
+                self._samples[self._key] = self._value
+                self._samples.sync()
+
+        def set(self, value):
+            with self._lock:
+                self._value = value
+                self._samples[self._key] = self._value
+                self._samples.sync()
+
+        def get(self):
+            with self._lock:
+                return self._value
+
+    return _ShelveValue
+
+
+# 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 enviroment variable.
+if 'prometheus_multiproc_dir' in os.environ:
+    _ValueClass = _MultiProcessValue()
+else:
+    _ValueClass = _MutexValue
 
 
 class _LabelWrapper(object):
@@ -387,7 +445,7 @@ def f():
     _reserved_labelnames = []
 
     def __init__(self, name, labelnames, labelvalues):
-        self._value = _ValueClass(name, labelnames, labelvalues)
+        self._value = _ValueClass(self._type, name, name, labelnames, labelvalues)
 
     def inc(self, amount=1):
         '''Increment counter by the given amount.'''
@@ -449,8 +507,12 @@ def f():
     _type = 'gauge'
     _reserved_labelnames = []
 
-    def __init__(self, name, labelnames, labelvalues):
-        self._value = _ValueClass(name, labelnames, labelvalues)
+    def __init__(self, name, labelnames, labelvalues, multiprocess_mode='all'):
+        if (_ValueClass._multiprocess
+                and multiprocess_mode not in ['min', 'max', 'livesum', 'liveall', 'all']):
+            raise ValueError('Invalid multiprocess mode: ' + multiprocess_mode)
+        self._value = _ValueClass(self._type, name, name, labelnames,
+                labelvalues, multiprocess_mode=multiprocess_mode)
 
     def inc(self, amount=1):
         '''Increment gauge by the given amount.'''
@@ -533,8 +595,8 @@ def create_response(request):
     _reserved_labelnames = ['quantile']
 
     def __init__(self, name, labelnames, labelvalues):
-        self._count = _ValueClass(name + '_count', labelnames, labelvalues)
-        self._sum = _ValueClass(name + '_sum', labelnames, labelvalues)
+        self._count = _ValueClass(self._type, name, name + '_count', labelnames, labelvalues)
+        self._sum = _ValueClass(self._type, name, name + '_sum', labelnames, labelvalues)
 
     def observe(self, amount):
         '''Observe the given amount.'''
@@ -607,7 +669,7 @@ def create_response(request):
     _reserved_labelnames = ['histogram']
 
     def __init__(self, name, labelnames, labelvalues, buckets=(.005, .01, .025, .05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0, 7.5, 10.0, _INF)):
-        self._sum = _ValueClass(name + '_sum', labelnames, labelvalues)
+        self._sum = _ValueClass(self._type, name, name + '_sum', labelnames, labelvalues)
         buckets = [float(b) for b in buckets]
         if buckets != sorted(buckets):
             # This is probably an error on the part of the user,
@@ -621,7 +683,7 @@ def __init__(self, name, labelnames, labelvalues, buckets=(.005, .01, .025, .05,
         self._buckets = []
         bucket_labelnames = labelnames + ('le',)
         for b in buckets:
-          self._buckets.append(_ValueClass(name + '_bucket', bucket_labelnames, labelvalues + (_floatToGoString(b),)))
+          self._buckets.append(_ValueClass(self._type, name, name + '_bucket', bucket_labelnames, labelvalues + (_floatToGoString(b),)))
 
     def observe(self, amount):
         '''Observe the given amount.'''

prometheus_client/multiprocess.py

@@ -0,0 +1,92 @@
+#!/usr/bin/python
+
+from __future__ import unicode_literals
+
+import glob
+import json
+import os
+import shelve
+
+from . import core
+
+class MultiProcessCollector(object):
+    """Collector for files for multi-process mode."""
+    def __init__(self, registry, path=os.environ.get('prometheus_multiproc_dir')):
+        self._path = path
+        if registry:
+          registry.register(self)
+
+    def collect(self):
+        metrics = {}
+        for f in glob.glob(os.path.join(self._path, '*.db')):
+            parts = os.path.basename(f).split('_')
+            typ = parts[0]
+            for key, value in shelve.open(f).items():
+                metric_name, name, labelnames, labelvalues = json.loads(key)
+                metrics.setdefault(metric_name, core.Metric(metric_name, 'Multiprocess metric', typ))
+                metric = metrics[metric_name]
+                if typ == 'gauge':
+                    pid = parts[2][:-3]
+                    metric._multiprocess_mode = parts[1]
+                    metric.add_sample(name, tuple(zip(labelnames, labelvalues)) + (('pid', pid), ), value)
+                else:
+                    # The deplucates and labels are fixed in the next for.
+                    metric.add_sample(name, tuple(zip(labelnames, labelvalues)), value)
+
+        for metric in metrics.values():
+            samples = {}
+            buckets = {}
+            for name, labels, value in metric.samples:
+                if metric.type == 'gauge':
+                    without_pid = tuple([l for l in labels if l[0] != 'pid'])
+                    if metric._multiprocess_mode == 'min':
+                        samples.setdefault((name, without_pid), value)
+                        if samples[(name, without_pid)] > value:
+                            samples[(name, without_pid)] = value
+                    elif metric._multiprocess_mode == 'max':
+                        samples.setdefault((name, without_pid), value)
+                        if samples[(name, without_pid)] < value:
+                            samples[(name, without_pid)] = value
+                    elif metric._multiprocess_mode == 'livesum':
+                        samples.setdefault((name, without_pid), 0.0)
+                        samples[(name, without_pid)] += value
+                    else:  # all/liveall
+                        samples[(name, labels)] = value
+                elif metric.type == 'histogram':
+                    bucket = [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.setdefault((name, labels), 0.0)
+                        samples[(name, labels)] += value
+                else:
+                    # Counter and Summary.
+                    samples.setdefault((name, labels), 0.0)
+                    samples[(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()):
+                        acc += value
+                        samples[(metric.name + '_bucket', labels + (('le', core._floatToGoString(bucket)), ))] = acc
+                    samples[(metric.name + '_count', labels)] = acc
+
+            # Convert to correct sample format.
+            metric.samples = [(name, dict(labels), value) for (name, labels), value in samples.items()]
+        return metrics.values()
+
+
+def mark_process_dead(pid, path=os.environ.get('prometheus_multiproc_dir')):
+    """Do bookkeeping for when one process dies in a multi-process setup."""
+    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)