Support slashes in pgw grouping keys.

prometheus_client/exposition.py

@@ -343,7 +343,7 @@ def _use_gateway(method, gateway, job, registry, grouping_key, timeout, handler)
     gateway_url = urlparse(gateway)
     if not gateway_url.scheme or (PYTHON26_OR_OLDER and gateway_url.scheme not in ['http', 'https']):
         gateway = 'http://{0}'.format(gateway)
-    url = '{0}/metrics/job/{1}'.format(gateway, quote_plus(job))
+    url = '{0}/metrics/{1}/{2}'.format(gateway, *_escape_grouping_key("job", job))
 
     data = b''
     if method != 'DELETE':
@@ -352,7 +352,7 @@ def _use_gateway(method, gateway, job, registry, grouping_key, timeout, handler)
     if grouping_key is None:
         grouping_key = {}
     url += ''.join(
-        '/{0}/{1}'.format(quote_plus(str(k)), quote_plus(str(v)))
+        '/{0}/{1}'.format(*_escape_grouping_key(str(k), str(v)))
         for k, v in sorted(grouping_key.items()))
 
     handler(
@@ -361,6 +361,14 @@ def _use_gateway(method, gateway, job, registry, grouping_key, timeout, handler)
     )()
 
 
+def _escape_grouping_key(k, v):
+    if '/' in v:
+        # Added in Pushgateway 0.9.0.
+        return k + "@base64", base64.urlsafe_b64encode(v.encode("utf-8")).decode("utf-8")
+    else:
+        return k, quote_plus(v)
+
+
 def instance_ip_grouping_key():
     """Grouping key with instance set to the IP Address of this host."""
     with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as s:

tests/test_exposition.py

@@ -239,7 +239,14 @@ def test_push_with_groupingkey(self):
     def test_push_with_complex_groupingkey(self):
         push_to_gateway(self.address, "my_job", self.registry, {'a': 9, 'b': 'a/ z'})
         self.assertEqual(self.requests[0][0].command, 'PUT')
-        self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a/9/b/a%2F+z')
+        self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a/9/b@base64/YS8geg==')
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
+
+    def test_push_with_complex_job(self):
+        push_to_gateway(self.address, "my/job", self.registry)
+        self.assertEqual(self.requests[0][0].command, 'PUT')
+        self.assertEqual(self.requests[0][0].path, '/metrics/job@base64/bXkvam9i')
         self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')