Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import base64
2from contextlib import closing
3import gzip
4from http.server import BaseHTTPRequestHandler
5import os
6import socket
7from socketserver import ThreadingMixIn
8import ssl
9import sys
10import threading
11from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
12from urllib.error import HTTPError
13from urllib.parse import parse_qs, quote_plus, urlparse
14from urllib.request import (
15 BaseHandler, build_opener, HTTPHandler, HTTPRedirectHandler, HTTPSHandler,
16 Request,
17)
18from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer
20from .openmetrics import exposition as openmetrics
21from .registry import CollectorRegistry, REGISTRY
22from .utils import floatToGoString
24__all__ = (
25 'CONTENT_TYPE_LATEST',
26 'delete_from_gateway',
27 'generate_latest',
28 'instance_ip_grouping_key',
29 'make_asgi_app',
30 'make_wsgi_app',
31 'MetricsHandler',
32 'push_to_gateway',
33 'pushadd_to_gateway',
34 'start_http_server',
35 'start_wsgi_server',
36 'write_to_textfile',
37)
39CONTENT_TYPE_LATEST = 'text/plain; version=0.0.4; charset=utf-8'
40"""Content type of the latest text format"""
43class _PrometheusRedirectHandler(HTTPRedirectHandler):
44 """
45 Allow additional methods (e.g. PUT) and data forwarding in redirects.
47 Use of this class constitute a user's explicit agreement to the
48 redirect responses the Prometheus client will receive when using it.
49 You should only use this class if you control or otherwise trust the
50 redirect behavior involved and are certain it is safe to full transfer
51 the original request (method and data) to the redirected URL. For
52 example, if you know there is a cosmetic URL redirect in front of a
53 local deployment of a Prometheus server, and all redirects are safe,
54 this is the class to use to handle redirects in that case.
56 The standard HTTPRedirectHandler does not forward request data nor
57 does it allow redirected PUT requests (which Prometheus uses for some
58 operations, for example `push_to_gateway`) because these cannot
59 generically guarantee no violations of HTTP RFC 2616 requirements for
60 the user to explicitly confirm redirects that could have unexpected
61 side effects (such as rendering a PUT request non-idempotent or
62 creating multiple resources not named in the original request).
63 """
65 def redirect_request(self, req, fp, code, msg, headers, newurl):
66 """
67 Apply redirect logic to a request.
69 See parent HTTPRedirectHandler.redirect_request for parameter info.
71 If the redirect is disallowed, this raises the corresponding HTTP error.
72 If the redirect can't be determined, return None to allow other handlers
73 to try. If the redirect is allowed, return the new request.
75 This method specialized for the case when (a) the user knows that the
76 redirect will not cause unacceptable side effects for any request method,
77 and (b) the user knows that any request data should be passed through to
78 the redirect. If either condition is not met, this should not be used.
79 """
80 # note that requests being provided by a handler will use get_method to
81 # indicate the method, by monkeypatching this, instead of setting the
82 # Request object's method attribute.
83 m = getattr(req, "method", req.get_method())
84 if not (code in (301, 302, 303, 307) and m in ("GET", "HEAD")
85 or code in (301, 302, 303) and m in ("POST", "PUT")):
86 raise HTTPError(req.full_url, code, msg, headers, fp)
87 new_request = Request(
88 newurl.replace(' ', '%20'), # space escaping in new url if needed.
89 headers=req.headers,
90 origin_req_host=req.origin_req_host,
91 unverifiable=True,
92 data=req.data,
93 )
94 new_request.method = m
95 return new_request
98def _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression):
99 """Bake output for metrics output."""
100 # Choose the correct plain text format of the output.
101 encoder, content_type = choose_encoder(accept_header)
102 if 'name[]' in params:
103 registry = registry.restricted_registry(params['name[]'])
104 output = encoder(registry)
105 headers = [('Content-Type', content_type)]
106 # If gzip encoding required, gzip the output.
107 if not disable_compression and gzip_accepted(accept_encoding_header):
108 output = gzip.compress(output)
109 headers.append(('Content-Encoding', 'gzip'))
110 return '200 OK', headers, output
113def make_wsgi_app(registry: CollectorRegistry = REGISTRY, disable_compression: bool = False) -> Callable:
114 """Create a WSGI app which serves the metrics from a registry."""
116 def prometheus_app(environ, start_response):
117 # Prepare parameters
118 accept_header = environ.get('HTTP_ACCEPT')
119 accept_encoding_header = environ.get('HTTP_ACCEPT_ENCODING')
120 params = parse_qs(environ.get('QUERY_STRING', ''))
121 method = environ['REQUEST_METHOD']
123 if method == 'OPTIONS':
124 status = '200 OK'
125 headers = [('Allow', 'OPTIONS,GET')]
126 output = b''
127 elif method != 'GET':
128 status = '405 Method Not Allowed'
129 headers = [('Allow', 'OPTIONS,GET')]
130 output = '# HTTP {}: {}; use OPTIONS or GET\n'.format(status, method).encode()
131 elif environ['PATH_INFO'] == '/favicon.ico':
132 # Serve empty response for browsers
133 status = '200 OK'
134 headers = [('', '')]
135 output = b''
136 else:
137 # Note: For backwards compatibility, the URI path for GET is not
138 # constrained to the documented /metrics, but any path is allowed.
139 # Bake output
140 status, headers, output = _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression)
141 # Return output
142 start_response(status, headers)
143 return [output]
145 return prometheus_app
148class _SilentHandler(WSGIRequestHandler):
149 """WSGI handler that does not log requests."""
151 def log_message(self, format, *args):
152 """Log nothing."""
155class ThreadingWSGIServer(ThreadingMixIn, WSGIServer):
156 """Thread per request HTTP server."""
157 # Make worker threads "fire and forget". Beginning with Python 3.7 this
158 # prevents a memory leak because ``ThreadingMixIn`` starts to gather all
159 # non-daemon threads in a list in order to join on them at server close.
160 daemon_threads = True
163def _get_best_family(address, port):
164 """Automatically select address family depending on address"""
165 # HTTPServer defaults to AF_INET, which will not start properly if
166 # binding an ipv6 address is requested.
167 # This function is based on what upstream python did for http.server
168 # in https://github.com/python/cpython/pull/11767
169 infos = socket.getaddrinfo(address, port, type=socket.SOCK_STREAM, flags=socket.AI_PASSIVE)
170 family, _, _, _, sockaddr = next(iter(infos))
171 return family, sockaddr[0]
174def _get_ssl_ctx(
175 certfile: str,
176 keyfile: str,
177 protocol: int,
178 cafile: Optional[str] = None,
179 capath: Optional[str] = None,
180 client_auth_required: bool = False,
181) -> ssl.SSLContext:
182 """Load context supports SSL."""
183 ssl_cxt = ssl.SSLContext(protocol=protocol)
185 if cafile is not None or capath is not None:
186 try:
187 ssl_cxt.load_verify_locations(cafile, capath)
188 except IOError as exc:
189 exc_type = type(exc)
190 msg = str(exc)
191 raise exc_type(f"Cannot load CA certificate chain from file "
192 f"{cafile!r} or directory {capath!r}: {msg}")
193 else:
194 try:
195 ssl_cxt.load_default_certs(purpose=ssl.Purpose.CLIENT_AUTH)
196 except IOError as exc:
197 exc_type = type(exc)
198 msg = str(exc)
199 raise exc_type(f"Cannot load default CA certificate chain: {msg}")
201 if client_auth_required:
202 ssl_cxt.verify_mode = ssl.CERT_REQUIRED
204 try:
205 ssl_cxt.load_cert_chain(certfile=certfile, keyfile=keyfile)
206 except IOError as exc:
207 exc_type = type(exc)
208 msg = str(exc)
209 raise exc_type(f"Cannot load server certificate file {certfile!r} or "
210 f"its private key file {keyfile!r}: {msg}")
212 return ssl_cxt
215def start_wsgi_server(
216 port: int,
217 addr: str = '0.0.0.0',
218 registry: CollectorRegistry = REGISTRY,
219 certfile: Optional[str] = None,
220 keyfile: Optional[str] = None,
221 client_cafile: Optional[str] = None,
222 client_capath: Optional[str] = None,
223 protocol: int = ssl.PROTOCOL_TLS_SERVER,
224 client_auth_required: bool = False,
225) -> Tuple[WSGIServer, threading.Thread]:
226 """Starts a WSGI server for prometheus metrics as a daemon thread."""
228 class TmpServer(ThreadingWSGIServer):
229 """Copy of ThreadingWSGIServer to update address_family locally"""
231 TmpServer.address_family, addr = _get_best_family(addr, port)
232 app = make_wsgi_app(registry)
233 httpd = make_server(addr, port, app, TmpServer, handler_class=_SilentHandler)
234 if certfile and keyfile:
235 context = _get_ssl_ctx(certfile, keyfile, protocol, client_cafile, client_capath, client_auth_required)
236 httpd.socket = context.wrap_socket(httpd.socket, server_side=True)
237 t = threading.Thread(target=httpd.serve_forever)
238 t.daemon = True
239 t.start()
241 return httpd, t
244start_http_server = start_wsgi_server
247def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
248 """Returns the metrics from the registry in latest text format as a string."""
250 def sample_line(line):
251 if line.labels:
252 labelstr = '{{{0}}}'.format(','.join(
253 ['{}="{}"'.format(
254 k, v.replace('\\', r'\\').replace('\n', r'\n').replace('"', r'\"'))
255 for k, v in sorted(line.labels.items())]))
256 else:
257 labelstr = ''
258 timestamp = ''
259 if line.timestamp is not None:
260 # Convert to milliseconds.
261 timestamp = f' {int(float(line.timestamp) * 1000):d}'
262 return f'{line.name}{labelstr} {floatToGoString(line.value)}{timestamp}\n'
264 output = []
265 for metric in registry.collect():
266 try:
267 mname = metric.name
268 mtype = metric.type
269 # Munging from OpenMetrics into Prometheus format.
270 if mtype == 'counter':
271 mname = mname + '_total'
272 elif mtype == 'info':
273 mname = mname + '_info'
274 mtype = 'gauge'
275 elif mtype == 'stateset':
276 mtype = 'gauge'
277 elif mtype == 'gaugehistogram':
278 # A gauge histogram is really a gauge,
279 # but this captures the structure better.
280 mtype = 'histogram'
281 elif mtype == 'unknown':
282 mtype = 'untyped'
284 output.append('# HELP {} {}\n'.format(
285 mname, metric.documentation.replace('\\', r'\\').replace('\n', r'\n')))
286 output.append(f'# TYPE {mname} {mtype}\n')
288 om_samples: Dict[str, List[str]] = {}
289 for s in metric.samples:
290 for suffix in ['_created', '_gsum', '_gcount']:
291 if s.name == metric.name + suffix:
292 # OpenMetrics specific sample, put in a gauge at the end.
293 om_samples.setdefault(suffix, []).append(sample_line(s))
294 break
295 else:
296 output.append(sample_line(s))
297 except Exception as exception:
298 exception.args = (exception.args or ('',)) + (metric,)
299 raise
301 for suffix, lines in sorted(om_samples.items()):
302 output.append('# HELP {}{} {}\n'.format(metric.name, suffix,
303 metric.documentation.replace('\\', r'\\').replace('\n', r'\n')))
304 output.append(f'# TYPE {metric.name}{suffix} gauge\n')
305 output.extend(lines)
306 return ''.join(output).encode('utf-8')
309def choose_encoder(accept_header: str) -> Tuple[Callable[[CollectorRegistry], bytes], str]:
310 accept_header = accept_header or ''
311 for accepted in accept_header.split(','):
312 if accepted.split(';')[0].strip() == 'application/openmetrics-text':
313 return (openmetrics.generate_latest,
314 openmetrics.CONTENT_TYPE_LATEST)
315 return generate_latest, CONTENT_TYPE_LATEST
318def gzip_accepted(accept_encoding_header: str) -> bool:
319 accept_encoding_header = accept_encoding_header or ''
320 for accepted in accept_encoding_header.split(','):
321 if accepted.split(';')[0].strip().lower() == 'gzip':
322 return True
323 return False
326class MetricsHandler(BaseHTTPRequestHandler):
327 """HTTP handler that gives metrics from ``REGISTRY``."""
328 registry: CollectorRegistry = REGISTRY
330 def do_GET(self) -> None:
331 # Prepare parameters
332 registry = self.registry
333 accept_header = self.headers.get('Accept')
334 accept_encoding_header = self.headers.get('Accept-Encoding')
335 params = parse_qs(urlparse(self.path).query)
336 # Bake output
337 status, headers, output = _bake_output(registry, accept_header, accept_encoding_header, params, False)
338 # Return output
339 self.send_response(int(status.split(' ')[0]))
340 for header in headers:
341 self.send_header(*header)
342 self.end_headers()
343 self.wfile.write(output)
345 def log_message(self, format: str, *args: Any) -> None:
346 """Log nothing."""
348 @classmethod
349 def factory(cls, registry: CollectorRegistry) -> type:
350 """Returns a dynamic MetricsHandler class tied
351 to the passed registry.
352 """
353 # This implementation relies on MetricsHandler.registry
354 # (defined above and defaulted to REGISTRY).
356 # As we have unicode_literals, we need to create a str()
357 # object for type().
358 cls_name = str(cls.__name__)
359 MyMetricsHandler = type(cls_name, (cls, object),
360 {"registry": registry})
361 return MyMetricsHandler
364def write_to_textfile(path: str, registry: CollectorRegistry) -> None:
365 """Write metrics to the given path.
367 This is intended for use with the Node exporter textfile collector.
368 The path must end in .prom for the textfile collector to process it."""
369 tmppath = f'{path}.{os.getpid()}.{threading.current_thread().ident}'
370 with open(tmppath, 'wb') as f:
371 f.write(generate_latest(registry))
373 # rename(2) is atomic but fails on Windows if the destination file exists
374 if os.name == 'nt':
375 os.replace(tmppath, path)
376 else:
377 os.rename(tmppath, path)
380def _make_handler(
381 url: str,
382 method: str,
383 timeout: Optional[float],
384 headers: Sequence[Tuple[str, str]],
385 data: bytes,
386 base_handler: Union[BaseHandler, type],
387) -> Callable[[], None]:
388 def handle() -> None:
389 request = Request(url, data=data)
390 request.get_method = lambda: method # type: ignore
391 for k, v in headers:
392 request.add_header(k, v)
393 resp = build_opener(base_handler).open(request, timeout=timeout)
394 if resp.code >= 400:
395 raise OSError(f"error talking to pushgateway: {resp.code} {resp.msg}")
397 return handle
400def default_handler(
401 url: str,
402 method: str,
403 timeout: Optional[float],
404 headers: List[Tuple[str, str]],
405 data: bytes,
406) -> Callable[[], None]:
407 """Default handler that implements HTTP/HTTPS connections.
409 Used by the push_to_gateway functions. Can be re-used by other handlers."""
411 return _make_handler(url, method, timeout, headers, data, HTTPHandler)
414def passthrough_redirect_handler(
415 url: str,
416 method: str,
417 timeout: Optional[float],
418 headers: List[Tuple[str, str]],
419 data: bytes,
420) -> Callable[[], None]:
421 """
422 Handler that automatically trusts redirect responses for all HTTP methods.
424 Augments standard HTTPRedirectHandler capability by permitting PUT requests,
425 preserving the method upon redirect, and passing through all headers and
426 data from the original request. Only use this handler if you control or
427 trust the source of redirect responses you encounter when making requests
428 via the Prometheus client. This handler will simply repeat the identical
429 request, including same method and data, to the new redirect URL."""
431 return _make_handler(url, method, timeout, headers, data, _PrometheusRedirectHandler)
434def basic_auth_handler(
435 url: str,
436 method: str,
437 timeout: Optional[float],
438 headers: List[Tuple[str, str]],
439 data: bytes,
440 username: Optional[str] = None,
441 password: Optional[str] = None,
442) -> Callable[[], None]:
443 """Handler that implements HTTP/HTTPS connections with Basic Auth.
445 Sets auth headers using supplied 'username' and 'password', if set.
446 Used by the push_to_gateway functions. Can be re-used by other handlers."""
448 def handle():
449 """Handler that implements HTTP Basic Auth.
450 """
451 if username is not None and password is not None:
452 auth_value = f'{username}:{password}'.encode()
453 auth_token = base64.b64encode(auth_value)
454 auth_header = b'Basic ' + auth_token
455 headers.append(('Authorization', auth_header))
456 default_handler(url, method, timeout, headers, data)()
458 return handle
461def tls_auth_handler(
462 url: str,
463 method: str,
464 timeout: Optional[float],
465 headers: List[Tuple[str, str]],
466 data: bytes,
467 certfile: str,
468 keyfile: str,
469 cafile: Optional[str] = None,
470 protocol: int = ssl.PROTOCOL_TLS_CLIENT,
471 insecure_skip_verify: bool = False,
472) -> Callable[[], None]:
473 """Handler that implements an HTTPS connection with TLS Auth.
475 The default protocol (ssl.PROTOCOL_TLS_CLIENT) will also enable
476 ssl.CERT_REQUIRED and SSLContext.check_hostname by default. This can be
477 disabled by setting insecure_skip_verify to True.
479 Both this handler and the TLS feature on pushgateay are experimental."""
480 context = ssl.SSLContext(protocol=protocol)
481 if cafile is not None:
482 context.load_verify_locations(cafile)
483 else:
484 context.load_default_certs()
486 if insecure_skip_verify:
487 context.check_hostname = False
488 context.verify_mode = ssl.CERT_NONE
490 context.load_cert_chain(certfile=certfile, keyfile=keyfile)
491 handler = HTTPSHandler(context=context)
492 return _make_handler(url, method, timeout, headers, data, handler)
495def push_to_gateway(
496 gateway: str,
497 job: str,
498 registry: CollectorRegistry,
499 grouping_key: Optional[Dict[str, Any]] = None,
500 timeout: Optional[float] = 30,
501 handler: Callable = default_handler,
502) -> None:
503 """Push metrics to the given pushgateway.
505 `gateway` the url for your push gateway. Either of the form
506 'http://pushgateway.local', or 'pushgateway.local'.
507 Scheme defaults to 'http' if none is provided
508 `job` is the job label to be attached to all pushed metrics
509 `registry` is an instance of CollectorRegistry
510 `grouping_key` please see the pushgateway documentation for details.
511 Defaults to None
512 `timeout` is how long push will attempt to connect before giving up.
513 Defaults to 30s, can be set to None for no timeout.
514 `handler` is an optional function which can be provided to perform
515 requests to the 'gateway'.
516 Defaults to None, in which case an http or https request
517 will be carried out by a default handler.
518 If not None, the argument must be a function which accepts
519 the following arguments:
520 url, method, timeout, headers, and content
521 May be used to implement additional functionality not
522 supported by the built-in default handler (such as SSL
523 client certicates, and HTTP authentication mechanisms).
524 'url' is the URL for the request, the 'gateway' argument
525 described earlier will form the basis of this URL.
526 'method' is the HTTP method which should be used when
527 carrying out the request.
528 'timeout' requests not successfully completed after this
529 many seconds should be aborted. If timeout is None, then
530 the handler should not set a timeout.
531 'headers' is a list of ("header-name","header-value") tuples
532 which must be passed to the pushgateway in the form of HTTP
533 request headers.
534 The function should raise an exception (e.g. IOError) on
535 failure.
536 'content' is the data which should be used to form the HTTP
537 Message Body.
539 This overwrites all metrics with the same job and grouping_key.
540 This uses the PUT HTTP method."""
541 _use_gateway('PUT', gateway, job, registry, grouping_key, timeout, handler)
544def pushadd_to_gateway(
545 gateway: str,
546 job: str,
547 registry: Optional[CollectorRegistry],
548 grouping_key: Optional[Dict[str, Any]] = None,
549 timeout: Optional[float] = 30,
550 handler: Callable = default_handler,
551) -> None:
552 """PushAdd metrics to the given pushgateway.
554 `gateway` the url for your push gateway. Either of the form
555 'http://pushgateway.local', or 'pushgateway.local'.
556 Scheme defaults to 'http' if none is provided
557 `job` is the job label to be attached to all pushed metrics
558 `registry` is an instance of CollectorRegistry
559 `grouping_key` please see the pushgateway documentation for details.
560 Defaults to None
561 `timeout` is how long push will attempt to connect before giving up.
562 Defaults to 30s, can be set to None for no timeout.
563 `handler` is an optional function which can be provided to perform
564 requests to the 'gateway'.
565 Defaults to None, in which case an http or https request
566 will be carried out by a default handler.
567 See the 'prometheus_client.push_to_gateway' documentation
568 for implementation requirements.
570 This replaces metrics with the same name, job and grouping_key.
571 This uses the POST HTTP method."""
572 _use_gateway('POST', gateway, job, registry, grouping_key, timeout, handler)
575def delete_from_gateway(
576 gateway: str,
577 job: str,
578 grouping_key: Optional[Dict[str, Any]] = None,
579 timeout: Optional[float] = 30,
580 handler: Callable = default_handler,
581) -> None:
582 """Delete metrics from the given pushgateway.
584 `gateway` the url for your push gateway. Either of the form
585 'http://pushgateway.local', or 'pushgateway.local'.
586 Scheme defaults to 'http' if none is provided
587 `job` is the job label to be attached to all pushed metrics
588 `grouping_key` please see the pushgateway documentation for details.
589 Defaults to None
590 `timeout` is how long delete will attempt to connect before giving up.
591 Defaults to 30s, can be set to None for no timeout.
592 `handler` is an optional function which can be provided to perform
593 requests to the 'gateway'.
594 Defaults to None, in which case an http or https request
595 will be carried out by a default handler.
596 See the 'prometheus_client.push_to_gateway' documentation
597 for implementation requirements.
599 This deletes metrics with the given job and grouping_key.
600 This uses the DELETE HTTP method."""
601 _use_gateway('DELETE', gateway, job, None, grouping_key, timeout, handler)
604def _use_gateway(
605 method: str,
606 gateway: str,
607 job: str,
608 registry: Optional[CollectorRegistry],
609 grouping_key: Optional[Dict[str, Any]],
610 timeout: Optional[float],
611 handler: Callable,
612) -> None:
613 gateway_url = urlparse(gateway)
614 # See https://bugs.python.org/issue27657 for details on urlparse in py>=3.7.6.
615 if not gateway_url.scheme or gateway_url.scheme not in ['http', 'https']:
616 gateway = f'http://{gateway}'
618 gateway = gateway.rstrip('/')
619 url = '{}/metrics/{}/{}'.format(gateway, *_escape_grouping_key("job", job))
621 data = b''
622 if method != 'DELETE':
623 if registry is None:
624 registry = REGISTRY
625 data = generate_latest(registry)
627 if grouping_key is None:
628 grouping_key = {}
629 url += ''.join(
630 '/{}/{}'.format(*_escape_grouping_key(str(k), str(v)))
631 for k, v in sorted(grouping_key.items()))
633 handler(
634 url=url, method=method, timeout=timeout,
635 headers=[('Content-Type', CONTENT_TYPE_LATEST)], data=data,
636 )()
639def _escape_grouping_key(k, v):
640 if v == "":
641 # Per https://github.com/prometheus/pushgateway/pull/346.
642 return k + "@base64", "="
643 elif '/' in v:
644 # Added in Pushgateway 0.9.0.
645 return k + "@base64", base64.urlsafe_b64encode(v.encode("utf-8")).decode("utf-8")
646 else:
647 return k, quote_plus(v)
650def instance_ip_grouping_key() -> Dict[str, Any]:
651 """Grouping key with instance set to the IP Address of this host."""
652 with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as s:
653 if sys.platform == 'darwin':
654 # This check is done this way only on MacOS devices
655 # it is done this way because the localhost method does
656 # not work.
657 # This method was adapted from this StackOverflow answer:
658 # https://stackoverflow.com/a/28950776
659 s.connect(('10.255.255.255', 1))
660 else:
661 s.connect(('localhost', 0))
663 return {'instance': s.getsockname()[0]}
666from .asgi import make_asgi_app # noqa