1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
#!/usr/bin/python
from __future__ import unicode_literals
import os
import socket
import time
import threading
from contextlib import closing
from wsgiref.simple_server import make_server
from . import core
try:
from BaseHTTPServer import BaseHTTPRequestHandler
from BaseHTTPServer import HTTPServer
from urllib2 import build_opener, Request, HTTPHandler
from urllib import quote_plus
from urlparse import parse_qs, urlparse
except ImportError:
# Python 3
unicode = str
from http.server import BaseHTTPRequestHandler
from http.server import HTTPServer
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'''
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
def start_wsgi_server(port, addr='', registry=core.REGISTRY):
"""Starts a WSGI server for prometheus metrics as a daemon thread."""
class PrometheusMetricsServer(threading.Thread):
def run(self):
httpd = make_server(addr, port, make_wsgi_app(registry))
httpd.serve_forever()
t = PrometheusMetricsServer()
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):
def do_GET(self):
registry = core.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):
return
def start_http_server(port, addr=''):
"""Starts a HTTP server for prometheus metrics as a daemon thread."""
class PrometheusMetricsServer(threading.Thread):
def run(self):
httpd = HTTPServer((addr, port), MetricsHandler)
httpd.serve_forever()
t = PrometheusMetricsServer()
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 push_to_gateway(gateway, job, registry, grouping_key=None, timeout=None):
'''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
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)
def pushadd_to_gateway(gateway, job, registry, grouping_key=None, timeout=None):
'''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
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)
def delete_from_gateway(gateway, job, grouping_key=None, timeout=None):
'''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
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)
def _use_gateway(method, gateway, job, registry, grouping_key, timeout):
if not (gateway.startswith('http://') or gateway.startswith('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())])
request = Request(url, data=data)
request.add_header('Content-Type', CONTENT_TYPE_LATEST)
request.get_method = lambda: method
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))
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]}
|
