Repository: prometheus/client_python
Pull request #169 (2017-06-15 16:16:26)

Reset multiproc values when the pid changes.
This puts us at ~2000ns for an operation,
which is ~300ns than previously but still slightly
better than we were before we switched to a single lock.

This is twice as slow as without multiproc mode.


Patch:
diff --git a/README.md b/README.md
index 9e3c5f2..8aebf65 100644
--- a/README.md
+++ b/README.md
@@ -416,7 +416,6 @@ This comes with a number of limitations:
 - Custom collectors do not work (e.g. cpu and memory metrics)
 - The pushgateway cannot be used
 - Gauges cannot use the `pid` label
-- Gunicorn's `preload_app` feature and equivalents are not supported
 
 There's several steps to getting this working:
 
@@ -480,7 +479,7 @@ system.
 
 ```python
 from prometheus_client.parser import text_string_to_metric_families
-for family in text_string_to_metric_families("my_gauge 1.0\n"):
+for family in text_string_to_metric_families(u"my_gauge 1.0\n"):
   for sample in family.samples:
     print("Name: {0} Labels: {1} Value: {2}".format(*sample))
 ```
diff --git a/prometheus_client/__init__.py b/prometheus_client/__init__.py
index b270a86..9aa4cdf 100644
--- a/prometheus_client/__init__.py
+++ b/prometheus_client/__init__.py
@@ -6,8 +6,6 @@ from . import process_collector
 from . import platform_collector
 
 __all__ = ['Counter', 'Gauge', 'Summary', 'Histogram']
-# http://stackoverflow.com/questions/19913653/no-unicode-in-all-for-a-packages-init
-__all__ = [n.encode('ascii') for n in __all__]
 
 CollectorRegistry = core.CollectorRegistry
 REGISTRY = core.REGISTRY
diff --git a/prometheus_client/bridge/graphite.py b/prometheus_client/bridge/graphite.py
index 4c71554..6b8f241 100755
--- a/prometheus_client/bridge/graphite.py
+++ b/prometheus_client/bridge/graphite.py
@@ -45,7 +45,7 @@ class _RegularPush(threading.Thread):
 
 
 class GraphiteBridge(object):
-    def __init__(self, address, registry=core.REGISTRY, timeout_seconds=30, _timer=default_timer):
+    def __init__(self, address, registry=core.REGISTRY, timeout_seconds=30, _timer=time.time):
         self._address = address
         self._registry = registry
         self._timeout = timeout_seconds
@@ -64,8 +64,8 @@ class GraphiteBridge(object):
                 if labels:
                     labelstr = '.' + '.'.join(
                         ['{0}.{1}'.format(
-                             _sanitize(k), _sanitize(v))
-                             for k, v in sorted(labels.items())])
+                            _sanitize(k), _sanitize(v))
+                            for k, v in sorted(labels.items())])
                 else:
                     labelstr = ''
                 output.append('{0}{1}{2} {3} {4}\n'.format(
diff --git a/prometheus_client/core.py b/prometheus_client/core.py
index d715af3..6456749 100644
--- a/prometheus_client/core.py
+++ b/prometheus_client/core.py
@@ -392,9 +392,9 @@ class _MmapedDict(object):
             self._f = None
 
 
-def _MultiProcessValue(__pid=os.getpid()):
-    pid = __pid
+def _MultiProcessValue(_pidFunc=os.getpid):
     files = {}
+    values = []
     # Use a single global lock when in multi-processing mode
     # as we presume this means there is no threading going on.
     # This avoids the need to also have mutexes in __MmapDict.
@@ -406,31 +406,51 @@ def _MultiProcessValue(__pid=os.getpid()):
         _multiprocess = True
 
         def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', **kwargs):
+            self._params = typ, metric_name, name, labelnames, labelvalues, multiprocess_mode
+            with lock:
+                self.__reset()
+                values.append(self)
+
+
+        def __reset(self):
+            self._pid = _pidFunc()
+            typ, metric_name, name, labelnames, labelvalues, multiprocess_mode = self._params
             if typ == 'gauge':
                 file_prefix = typ + '_' +  multiprocess_mode
             else:
                 file_prefix = typ
-            with lock:
-                if file_prefix not in files:
-                    filename = os.path.join(
-                            os.environ['prometheus_multiproc_dir'], '{0}_{1}.db'.format(file_prefix, pid))
-                    files[file_prefix] = _MmapedDict(filename)
+            if file_prefix not in files:
+                filename = os.path.join(
+                        os.environ['prometheus_multiproc_dir'], '{0}_{1}.db'.format(file_prefix, self._pid))
+                files[file_prefix] = _MmapedDict(filename)
             self._file = files[file_prefix]
             self._key = json.dumps((metric_name, name, labelnames, labelvalues))
             self._value = self._file.read_value(self._key)
 
+        def __check_for_pid_change(self):
+            if self._pid != _pidFunc():
+                # There has been a fork(), reset all the values.
+                for f in files.values():
+                    f.close()
+                files.clear()
+                for value in values:
+                    value.__reset()
+
         def inc(self, amount):
             with lock:
+                self.__check_for_pid_change()
                 self._value += amount
                 self._file.write_value(self._key, self._value)
 
         def set(self, value):
             with lock:
+                self.__check_for_pid_change()
                 self._value = value
                 self._file.write_value(self._key, self._value)
 
         def get(self):
             with lock:
+                self.__check_for_pid_change()
                 return self._value
 
     return _MmapedValue
diff --git a/prometheus_client/parser.py b/prometheus_client/parser.py
index 18e1a31..c4f4c77 100644
--- a/prometheus_client/parser.py
+++ b/prometheus_client/parser.py
@@ -12,7 +12,7 @@ from . import core
 
 
 def text_string_to_metric_families(text):
-    """Parse Prometheus text format from a string.
+    """Parse Prometheus text format from a unicode string.
 
     See text_fd_to_metric_families.
     """


Test patch:
diff --git a/tests/test_multiprocess.py b/tests/test_multiprocess.py
index c05e457..d122dbb 100644
--- a/tests/test_multiprocess.py
+++ b/tests/test_multiprocess.py
@@ -13,7 +13,7 @@ class TestMultiProcess(unittest.TestCase):
     def setUp(self):
         self.tempdir = tempfile.mkdtemp()
         os.environ['prometheus_multiproc_dir'] = self.tempdir
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(123)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 123)
         self.registry = CollectorRegistry()
         MultiProcessCollector(self.registry, self.tempdir)
 
@@ -24,7 +24,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_counter_adds(self):
         c1 = Counter('c', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         c2 = Counter('c', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('c'))
         c1.inc(1)
@@ -33,7 +33,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_summary_adds(self):
         s1 = Summary('s', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         s2 = Summary('s', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('s_count'))
         self.assertEqual(0, self.registry.get_sample_value('s_sum'))
@@ -44,7 +44,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_histogram_adds(self):
         h1 = Histogram('h', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         h2 = Histogram('h', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('h_count'))
         self.assertEqual(0, self.registry.get_sample_value('h_sum'))
@@ -57,7 +57,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_gauge_all(self):
         g1 = Gauge('g', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'}))
         self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'}))
@@ -69,7 +69,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_gauge_liveall(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._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'}))
@@ -83,7 +83,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_gauge_min(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min')
         self.assertEqual(0, self.registry.get_sample_value('g'))
         g1.set(1)
@@ -92,7 +92,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_gauge_max(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max')
         self.assertEqual(0, self.registry.get_sample_value('g'))
         g1.set(1)
@@ -101,7 +101,7 @@ class TestMultiProcess(unittest.TestCase):
 
     def test_gauge_livesum(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum')
         self.assertEqual(0, self.registry.get_sample_value('g'))
         g1.set(1)
@@ -115,6 +115,20 @@ class TestMultiProcess(unittest.TestCase):
          c1.inc(1)
          self.assertEqual(1, self.registry.get_sample_value('ns_ss_c'))
 
+    def test_counter_across_forks(self):
+        pid = 0
+        def get_pid():
+            return pid
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(get_pid)
+        c1 = Counter('c', 'help', registry=None)
+        self.assertEqual(0, self.registry.get_sample_value('c'))
+        c1.inc(1)
+        c1.inc(1)
+        pid = 1
+        c1.inc(1)
+        self.assertEqual(3, self.registry.get_sample_value('c'))
+        self.assertEqual(1, c1._value.get())
+
 
 class TestMmapedDict(unittest.TestCase):
     def setUp(self):