Repository: prometheus/client_python
Pull request #794 (2022-04-07 03:27:26)

Add `sum`, `livemin`, and `livemax` multiprocess modes for `Gauge`s
Hi!

I work on a multiprocess Flask application and we recently found ourselves wanting the equivalent of the `max` multiprocessing mode, but only for live processes. I came here to poke around the code and also noticed this recent issue asking for a `sum` mode https://github.com/prometheus/client_python/issues/793 , so it seems like there's some interest in more multiprocess modes.

If I understand correctly, the core change needed to support this is actually around the deletion of the files when a process dies - files for `live*` metrics should be deleted, others should not. So I added the new modes with the same collection logic as their non-live counterparts and expanded the check in `mark_process_dead()` to automatically include any metric with `live` as part of its multiprocess mode name (this might be excessively fancy, but it seemed useful to reduce potential future changes).

I also added tests to cover these new modes based on the existing tests and updated the `README.md` to describe the new modes.

csmarchbanks: Thanks! I should have some time next week to give this a review.

Patch:
diff --git a/README.md b/README.md
index d9721a5..b25a55b 100644
--- a/README.md
+++ b/README.md
@@ -609,14 +609,17 @@ def child_exit(server, worker):
 
 **4. Metrics tuning (Gauge)**:
 
-When `Gauge` metrics are used, additional tuning needs to be performed.
+When `Gauge`s are used in multiprocess applications,
+you must decide how to handle the metrics reported by each process.
 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.
+- 'all': Default. Return a timeseries per process (alive or dead), labelled by the process's `pid` (the label is added internally).
+- 'min': Return a single timeseries that is the minimum of the values of all processes (alive or dead).
+- 'max': Return a single timeseries that is the maximum of the values of all processes (alive or dead).
+- 'sum': Return a single timeseries that is the sum of the values of all processes (alive or dead).
+
+Prepend 'live' to the beginning of the mode to return the same result but only considering living processes
+(e.g., 'liveall, 'livesum', 'livemax', 'livemin').
 
 ```python
 from prometheus_client import Gauge
diff --git a/prometheus_client/exposition.py b/prometheus_client/exposition.py
index 86a9be4..345b247 100644
--- a/prometheus_client/exposition.py
+++ b/prometheus_client/exposition.py
@@ -13,6 +13,7 @@ from urllib.parse import parse_qs, quote_plus, urlparse
 from urllib.request import (
     build_opener, HTTPHandler, HTTPRedirectHandler, Request,
 )
+import warnings
 from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer
 
 from .openmetrics import exposition as openmetrics
@@ -97,10 +98,10 @@ class _PrometheusRedirectHandler(HTTPRedirectHandler):
 def _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression):
     """Bake output for metrics output."""
     # Choose the correct plain text format of the output.
-    formatter, content_type = choose_formatter(accept_header)
+    encoder, content_type = choose_encoder(accept_header)
     if 'name[]' in params:
         registry = registry.restricted_registry(params['name[]'])
-    output = formatter(registry)
+    output = encoder(registry)
     headers = [('Content-Type', content_type)]
     # If gzip encoding required, gzip the output.
     if not disable_compression and gzip_accepted(accept_encoding_header):
@@ -237,7 +238,7 @@ def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
     return ''.join(output).encode('utf-8')
 
 
-def choose_formatter(accept_header: str) -> Tuple[Callable[[CollectorRegistry], bytes], str]:
+def choose_encoder(accept_header: str) -> Tuple[Callable[[CollectorRegistry], bytes], str]:
     accept_header = accept_header or ''
     for accepted in accept_header.split(','):
         if accepted.split(';')[0].strip() == 'application/openmetrics-text':
@@ -246,6 +247,15 @@ def choose_formatter(accept_header: str) -> Tuple[Callable[[CollectorRegistry],
     return generate_latest, CONTENT_TYPE_LATEST
 
 
+def choose_formatter(accept_header: str) -> Tuple[Callable[[CollectorRegistry], bytes], str]:
+    warnings.warn(
+        "choose_formatter is deprecated and will be removed in 0.15.0, please use choose_encoder instead",
+        DeprecationWarning,
+        stacklevel=2
+    )
+    return choose_encoder(accept_header)
+
+
 def gzip_accepted(accept_encoding_header: str) -> bool:
     accept_encoding_header = accept_encoding_header or ''
     for accepted in accept_encoding_header.split(','):
diff --git a/prometheus_client/metrics.py b/prometheus_client/metrics.py
index 41bec51..8878fb8 100644
--- a/prometheus_client/metrics.py
+++ b/prometheus_client/metrics.py
@@ -346,7 +346,7 @@ class Gauge(MetricWrapperBase):
         d.set_function(lambda: len(my_dict))
     """
     _type = 'gauge'
-    _MULTIPROC_MODES = frozenset(('min', 'max', 'livesum', 'liveall', 'all'))
+    _MULTIPROC_MODES = frozenset(('all', 'liveall', 'min', 'livemin', 'max', 'livemax', 'sum', 'livesum'))
 
     def __init__(self,
                  name: str,
diff --git a/prometheus_client/multiprocess.py b/prometheus_client/multiprocess.py
index 6c95374..5a23c48 100644
--- a/prometheus_client/multiprocess.py
+++ b/prometheus_client/multiprocess.py
@@ -4,6 +4,7 @@ import json
 import os
 import warnings
 
+from .metrics import Gauge
 from .metrics_core import Metric
 from .mmap_dict import MmapedDict
 from .samples import Sample
@@ -63,8 +64,8 @@ class MultiProcessCollector:
             try:
                 file_values = MmapedDict.read_all_values_from_file(f)
             except FileNotFoundError:
-                if typ == 'gauge' and parts[1] in ('liveall', 'livesum'):
-                    # Those files can disappear between the glob of collect
+                if typ == 'gauge' and parts[1].startswith('live'):
+                    # Files for 'live*' gauges can be deleted between the glob of collect
                     # and now (via a mark_process_dead call) so don't fail if
                     # the file is missing
                     continue
@@ -96,15 +97,15 @@ class MultiProcessCollector:
                 name, labels, value, timestamp, exemplar = s
                 if metric.type == 'gauge':
                     without_pid_key = (name, tuple(l for l in labels if l[0] != 'pid'))
-                    if metric._multiprocess_mode == 'min':
+                    if metric._multiprocess_mode in ('min', 'livemin'):
                         current = samples_setdefault(without_pid_key, value)
                         if value < current:
                             samples[without_pid_key] = value
-                    elif metric._multiprocess_mode == 'max':
+                    elif metric._multiprocess_mode in ('max', 'livemax'):
                         current = samples_setdefault(without_pid_key, value)
                         if value > current:
                             samples[without_pid_key] = value
-                    elif metric._multiprocess_mode == 'livesum':
+                    elif metric._multiprocess_mode in ('sum', 'livesum'):
                         samples[without_pid_key] += value
                     else:  # all/liveall
                         samples[(name, labels)] = value
@@ -152,11 +153,13 @@ class MultiProcessCollector:
         return self.merge(files, accumulate=True)
 
 
+_LIVE_GAUGE_MULTIPROCESS_MODES = {m for m in Gauge._MULTIPROC_MODES if m.startswith('live')}
+
+
 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', os.environ.get('prometheus_multiproc_dir'))
-    for f in glob.glob(os.path.join(path, f'gauge_livesum_{pid}.db')):
-        os.remove(f)
-    for f in glob.glob(os.path.join(path, f'gauge_liveall_{pid}.db')):
-        os.remove(f)
+    for mode in _LIVE_GAUGE_MULTIPROCESS_MODES:
+        for f in glob.glob(os.path.join(path, f'gauge_{mode}_{pid}.db')):
+            os.remove(f)
diff --git a/setup.py b/setup.py
index 8b5841b..733b5e5 100644
--- a/setup.py
+++ b/setup.py
@@ -8,7 +8,7 @@ with open(path.join(path.abspath(path.dirname(__file__)), 'README.md')) as f:
 
 setup(
     name="prometheus_client",
-    version="0.14.0",
+    version="0.14.1",
     author="Brian Brazil",
     author_email="brian.brazil@robustperception.io",
     description="Python client for the Prometheus monitoring system.",


Test patch:
diff --git a/tests/test_exposition.py b/tests/test_exposition.py
index fd13055..8da12a1 100644
--- a/tests/test_exposition.py
+++ b/tests/test_exposition.py
@@ -2,6 +2,7 @@ from http.server import BaseHTTPRequestHandler, HTTPServer
 import threading
 import time
 import unittest
+import warnings
 
 import pytest
 
@@ -12,9 +13,10 @@ from prometheus_client import (
 )
 from prometheus_client.core import GaugeHistogramMetricFamily, Timestamp
 from prometheus_client.exposition import (
-    basic_auth_handler, default_handler, MetricsHandler,
-    passthrough_redirect_handler,
+    basic_auth_handler, choose_encoder, choose_formatter, default_handler,
+    MetricsHandler, passthrough_redirect_handler,
 )
+import prometheus_client.openmetrics.exposition as openmetrics
 
 
 class TestGenerateText(unittest.TestCase):
@@ -460,5 +462,19 @@ def test_histogram_metric_families(MetricFamily, registry, buckets, sum_value, e
     _expect_metric_exception(registry, error)
 
 
+def test_choose_encoder():
+    assert choose_encoder(None) == (generate_latest, CONTENT_TYPE_LATEST)
+    assert choose_encoder(CONTENT_TYPE_LATEST) == (generate_latest, CONTENT_TYPE_LATEST)
+    assert choose_encoder(openmetrics.CONTENT_TYPE_LATEST) == (openmetrics.generate_latest, openmetrics.CONTENT_TYPE_LATEST)
+
+
+def test_choose_formatter():
+    with warnings.catch_warnings(record=True) as w:
+        assert choose_formatter('') == (generate_latest, CONTENT_TYPE_LATEST)
+        assert len(w) == 1
+        assert issubclass(w[-1].category, DeprecationWarning)
+        assert "choose_formatter is deprecated" in str(w[-1].message)
+
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_multiprocess.py b/tests/test_multiprocess.py
index 9ec0578..a41903a 100644
--- a/tests/test_multiprocess.py
+++ b/tests/test_multiprocess.py
@@ -132,6 +132,17 @@ class TestMultiProcess(unittest.TestCase):
         g2.set(2)
         self.assertEqual(1, self.registry.get_sample_value('g'))
 
+    def test_gauge_livemin(self):
+        g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livemin')
+        values.ValueClass = MultiProcessValue(lambda: 456)
+        g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livemin')
+        self.assertEqual(0, self.registry.get_sample_value('g'))
+        g1.set(1)
+        g2.set(2)
+        self.assertEqual(1, self.registry.get_sample_value('g'))
+        mark_process_dead(123, os.environ['PROMETHEUS_MULTIPROC_DIR'])
+        self.assertEqual(2, self.registry.get_sample_value('g'))
+
     def test_gauge_max(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max')
         values.ValueClass = MultiProcessValue(lambda: 456)
@@ -141,6 +152,28 @@ class TestMultiProcess(unittest.TestCase):
         g2.set(2)
         self.assertEqual(2, self.registry.get_sample_value('g'))
 
+    def test_gauge_livemax(self):
+        g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livemax')
+        values.ValueClass = MultiProcessValue(lambda: 456)
+        g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livemax')
+        self.assertEqual(0, self.registry.get_sample_value('g'))
+        g1.set(2)
+        g2.set(1)
+        self.assertEqual(2, self.registry.get_sample_value('g'))
+        mark_process_dead(123, os.environ['PROMETHEUS_MULTIPROC_DIR'])
+        self.assertEqual(1, self.registry.get_sample_value('g'))
+
+    def test_gauge_sum(self):
+        g1 = Gauge('g', 'help', registry=None, multiprocess_mode='sum')
+        values.ValueClass = MultiProcessValue(lambda: 456)
+        g2 = Gauge('g', 'help', registry=None, multiprocess_mode='sum')
+        self.assertEqual(0, self.registry.get_sample_value('g'))
+        g1.set(1)
+        g2.set(2)
+        self.assertEqual(3, self.registry.get_sample_value('g'))
+        mark_process_dead(123, os.environ['PROMETHEUS_MULTIPROC_DIR'])
+        self.assertEqual(3, self.registry.get_sample_value('g'))
+
     def test_gauge_livesum(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum')
         values.ValueClass = MultiProcessValue(lambda: 456)