#!/usr/bin/python from __future__ import unicode_literals import base64 import os import socket import sys import threading from contextlib import closing from wsgiref.simple_server import make_server, WSGIRequestHandler from . import core try: from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer from SocketServer import ThreadingMixIn from urllib2 import build_opener, Request, HTTPHandler from urllib import quote_plus from urlparse import parse_qs, urlparse except ImportError: # Python 3 from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from urllib.request import build_opener, Request, HTTPHandler from urllib.parse import quote_plus, parse_qs, urlparse CONTENT_TYPE_LATEST = str('text/plain; version=0.0.4; charset=utf-8') '''Content type of the latest text format''' PYTHON26_OR_OLDER = sys.version_info < (2, 7) def make_wsgi_app(registry=core.REGISTRY): '''Create a WSGI app which serves the metrics from a registry.''' def prometheus_app(environ, start_response): params = parse_qs(environ['QUERY_STRING']) r = registry if 'name[]' in params: r = r.restricted_registry(params['name[]']) output = generate_latest(r) status = str('200 OK') headers = [(str('Content-type'), CONTENT_TYPE_LATEST)] start_response(status, headers) return [output] return prometheus_app class _SilentHandler(WSGIRequestHandler): """WSGI handler that does not log requests.""" def log_message(self, format, *args): """Log nothing.""" def start_wsgi_server(port, addr='', registry=core.REGISTRY): """Starts a WSGI server for prometheus metrics as a daemon thread.""" app = make_wsgi_app(registry) httpd = make_server(addr, port, app, handler_class=_SilentHandler) t = threading.Thread(target=httpd.serve_forever) t.daemon = True t.start() def generate_latest(registry=core.REGISTRY): '''Returns the metrics from the registry in latest text format as a string.''' output = [] for metric in registry.collect(): output.append('# HELP {0} {1}'.format( metric.name, metric.documentation.replace('\\', r'\\').replace('\n', r'\n'))) output.append('\n# TYPE {0} {1}\n'.format(metric.name, metric.type)) for name, labels, value in metric.samples: if labels: labelstr = '{{{0}}}'.format(','.join( ['{0}="{1}"'.format( k, v.replace('\\', r'\\').replace('\n', r'\n').replace('"', r'\"')) for k, v in sorted(labels.items())])) else: labelstr = '' output.append('{0}{1} {2}\n'.format(name, labelstr, core._floatToGoString(value))) return ''.join(output).encode('utf-8') class MetricsHandler(BaseHTTPRequestHandler): """HTTP handler that gives metrics from ``core.REGISTRY``.""" registry = core.REGISTRY def do_GET(self): registry = self.registry params = parse_qs(urlparse(self.path).query) if 'name[]' in params: registry = registry.restricted_registry(params['name[]']) try: output = generate_latest(registry) except: self.send_error(500, 'error generating metric output') raise self.send_response(200) self.send_header('Content-Type', CONTENT_TYPE_LATEST) self.end_headers() self.wfile.write(output) def log_message(self, format, *args): """Log nothing.""" @staticmethod def factory(registry): """Returns a dynamic MetricsHandler class tied to the passed registry. """ # This implementation relies on MetricsHandler.registry # (defined above and defaulted to core.REGISTRY). # As we have unicode_literals, we need to create a str() # object for type(). cls_name = str('MetricsHandler') MyMetricsHandler = type(cls_name, (MetricsHandler, object), {"registry": registry}) return MyMetricsHandler class _ThreadingSimpleServer(ThreadingMixIn, HTTPServer): """Thread per request HTTP server.""" def start_http_server(port, addr='', registry=core.REGISTRY): """Starts an HTTP server for prometheus metrics as a daemon thread""" CustomMetricsHandler = MetricsHandler.factory(registry) httpd = _ThreadingSimpleServer((addr, port), CustomMetricsHandler) t = threading.Thread(target=httpd.serve_forever) t.daemon = True t.start() def write_to_textfile(path, registry): '''Write metrics to the given path. This is intended for use with the Node exporter textfile collector. The path must end in .prom for the textfile collector to process it.''' tmppath = '%s.%s.%s' % (path, os.getpid(), threading.current_thread().ident) with open(tmppath, 'wb') as f: f.write(generate_latest(registry)) # rename(2) is atomic. os.rename(tmppath, path) def default_handler(url, method, timeout, headers, data): '''Default handler that implements HTTP/HTTPS connections. Used by the push_to_gateway functions. Can be re-used by other handlers.''' def handle(): request = Request(url, data=data) request.get_method = lambda: method for k, v in headers: request.add_header(k, v) resp = build_opener(HTTPHandler).open(request, timeout=timeout) if resp.code >= 400: raise IOError("error talking to pushgateway: {0} {1}".format( resp.code, resp.msg)) return handle def basic_auth_handler(url, method, timeout, headers, data, username=None, password=None): '''Handler that implements HTTP/HTTPS connections with Basic Auth. Sets auth headers using supplied 'username' and 'password', if set. Used by the push_to_gateway functions. Can be re-used by other handlers.''' def handle(): '''Handler that implements HTTP Basic Auth. ''' if username is not None and password is not None: auth_value = '{0}:{1}'.format(username, password).encode('utf-8') auth_token = base64.b64encode(auth_value) auth_header = b'Basic ' + auth_token headers.append(['Authorization', auth_header]) default_handler(url, method, timeout, headers, data)() return handle def push_to_gateway(gateway, job, registry, grouping_key=None, timeout=None, handler=default_handler): '''Push metrics to the given pushgateway. `gateway` the url for your push gateway. Either of the form 'http://pushgateway.local', or 'pushgateway.local'. Scheme defaults to 'http' if none is provided `job` is the job label to be attached to all pushed metrics `registry` is an instance of CollectorRegistry `grouping_key` please see the pushgateway documentation for details. Defaults to None `timeout` is how long push will attempt to connect before giving up. Defaults to None `handler` is an optional function which can be provided to perform requests to the 'gateway'. Defaults to None, in which case an http or https request will be carried out by a default handler. If not None, the argument must be a function which accepts the following arguments: url, method, timeout, headers, and content May be used to implement additional functionality not supported by the built-in default handler (such as SSL client certicates, and HTTP authentication mechanisms). 'url' is the URL for the request, the 'gateway' argument described earlier will form the basis of this URL. 'method' is the HTTP method which should be used when carrying out the request. 'timeout' requests not successfully completed after this many seconds should be aborted. If timeout is None, then the handler should not set a timeout. 'headers' is a list of ("header-name","header-value") tuples which must be passed to the pushgateway in the form of HTTP request headers. The function should raise an exception (e.g. IOError) on failure. 'content' is the data which should be used to form the HTTP Message Body. This overwrites all metrics with the same job and grouping_key. This uses the PUT HTTP method.''' _use_gateway('PUT', gateway, job, registry, grouping_key, timeout, handler) def pushadd_to_gateway(gateway, job, registry, grouping_key=None, timeout=None, handler=default_handler): '''PushAdd metrics to the given pushgateway. `gateway` the url for your push gateway. Either of the form 'http://pushgateway.local', or 'pushgateway.local'. Scheme defaults to 'http' if none is provided `job` is the job label to be attached to all pushed metrics `registry` is an instance of CollectorRegistry `grouping_key` please see the pushgateway documentation for details. Defaults to None `timeout` is how long push will attempt to connect before giving up. Defaults to None `handler` is an optional function which can be provided to perform requests to the 'gateway'. Defaults to None, in which case an http or https request will be carried out by a default handler. See the 'prometheus_client.push_to_gateway' documentation for implementation requirements. This replaces metrics with the same name, job and grouping_key. This uses the POST HTTP method.''' _use_gateway('POST', gateway, job, registry, grouping_key, timeout, handler) def delete_from_gateway(gateway, job, grouping_key=None, timeout=None, handler=default_handler): '''Delete metrics from the given pushgateway. `gateway` the url for your push gateway. Either of the form 'http://pushgateway.local', or 'pushgateway.local'. Scheme defaults to 'http' if none is provided `job` is the job label to be attached to all pushed metrics `grouping_key` please see the pushgateway documentation for details. Defaults to None `timeout` is how long delete will attempt to connect before giving up. Defaults to None `handler` is an optional function which can be provided to perform requests to the 'gateway'. Defaults to None, in which case an http or https request will be carried out by a default handler. See the 'prometheus_client.push_to_gateway' documentation for implementation requirements. This deletes metrics with the given job and grouping_key. This uses the DELETE HTTP method.''' _use_gateway('DELETE', gateway, job, None, grouping_key, timeout, handler) 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)) data = b'' if method != 'DELETE': data = generate_latest(registry) if grouping_key is None: grouping_key = {} url = url + ''.join(['/{0}/{1}'.format(quote_plus(str(k)), quote_plus(str(v))) for k, v in sorted(grouping_key.items())]) headers=[('Content-Type', CONTENT_TYPE_LATEST)] handler(url=url, method=method, timeout=timeout, headers=headers, data=data)() 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: s.connect(('localhost', 0)) return {'instance': s.getsockname()[0]}