We've forked the original prometheus python client for a few reasons, all stemming from multiprocessing handling
Postal-main uses a pretty standard gunicorn deployment (pre-fork web server). Glossing over many details: a gunicorn master process will fork worker processes to allow handling web requests in parallel. These workers process a few hundred requests, then exit, getting replaced by another forked worker. The use of processes, instead of threads, makes aggregating metrics from the workers less straightforward. What the mainline repo does, and what we do, is to use the filesystem to store the metrics for each of these workers. When prometheus scrapes the scrape endpoint, the client will read all the relevant metrics files, aggregate the results, then serve the metrics.
Likewise, our celery pods use --pool=prefork, and metrics need to be collected the same way
In a multiprocess setup, in which we treat multiple OS processes as one logical process (i.e. we are interested in the metrics for a pod, not each gunicorn worker in the pod), we need aggregate the metrics somehow. For counters and histograms, this is straightforward; we just sum up everything. For gauges, there are multiple strategies for aggregating these metrics, all equally valid. We might want to that the max value of all the process gauges for a high-water-mark, or we might want to, for a batch job that is fanned out to multiple workers, take a sum of all the gauges for live worker processes to track progress on that batch.
Many of our celery tasks have a pattern of querying the database to find entries which need to processed in some way, e.g. couriers which need to be paid out. Because of how we process this data, we needed a gauge aggregation strategy which just takes the latest gauge value, discarding the rest (search for "multiprocess_mode" below
Metrics files are identified by a pid. Since processes are constantly forked and exiting, this will, at best, generate a lot of metrics files, which all need to be opened for a scrape, and at worst, could have pid collisions. This fork runs a thread which goes and cleans up metrics files generated by exited processes, and merges them into an archive file
The official Python 2 and 3 client for Prometheus.
One: Install the client:
pip install prometheus_client
Two: Paste the following into a Python interpreter:
from prometheus_client import start_http_server, Summary
import random
import time
# Create a metric to track time spent and requests made.
REQUEST_TIME = Summary('request_processing_seconds', 'Time spent processing request')
# Decorate function with metric.
@REQUEST_TIME.time()
def process_request(t):
"""A dummy function that takes some time."""
time.sleep(t)
if __name__ == '__main__':
# Start up the server to expose the metrics.
start_http_server(8000)
# Generate some requests.
while True:
process_request(random.random())Three: Visit http://localhost:8000/ to view the metrics.
From one easy to use decorator you get:
request_processing_seconds_count: Number of times this function was called.request_processing_seconds_sum: Total amount of time spent in this function.
Prometheus's rate function allows calculation of both requests per second,
and latency over time from this data.
In addition if you're on Linux the process metrics expose CPU, memory and
other information about the process for free!
pip install prometheus_client
This package can be found on PyPI.
Four types of metric are offered: Counter, Gauge, Summary and Histogram. See the documentation on metric types and instrumentation best practices on how to use them.
Counters go up, and reset when the process restarts.
from prometheus_client import Counter
c = Counter('my_failures', 'Description of counter')
c.inc() # Increment by 1
c.inc(1.6) # Increment by given valueIf there is a suffix of _total on the metric name, it will be removed. When
exposing the time series for counter, a _total suffix will be added. This is
for compatibility between OpenMetrics and the Prometheus text format, as OpenMetrics
requires the _total suffix.
There are utilities to count exceptions raised:
@c.count_exceptions()
def f():
pass
with c.count_exceptions():
pass
# Count only one type of exception
with c.count_exceptions(ValueError):
passGauges can go up and down.
from prometheus_client import Gauge
g = Gauge('my_inprogress_requests', 'Description of gauge')
g.inc() # Increment by 1
g.dec(10) # Decrement by given value
g.set(4.2) # Set to a given valueThere are utilities for common use cases:
g.set_to_current_time() # Set to current unixtime
# Increment when entered, decrement when exited.
@g.track_inprogress()
def f():
pass
with g.track_inprogress():
passA Gauge can also take its value from a callback:
d = Gauge('data_objects', 'Number of objects')
my_dict = {}
d.set_function(lambda: len(my_dict))Summaries track the size and number of events.
from prometheus_client import Summary
s = Summary('request_latency_seconds', 'Description of summary')
s.observe(4.7) # Observe 4.7 (seconds in this case)There are utilities for timing code:
@s.time()
def f():
pass
with s.time():
passThe Python client doesn't store or expose quantile information at this time.
Histograms track the size and number of events in buckets. This allows for aggregatable calculation of quantiles.
from prometheus_client import Histogram
h = Histogram('request_latency_seconds', 'Description of histogram')
h.observe(4.7) # Observe 4.7 (seconds in this case)The default buckets are intended to cover a typical web/rpc request from milliseconds to seconds.
They can be overridden by passing buckets keyword argument to Histogram.
There are utilities for timing code:
@h.time()
def f():
pass
with h.time():
passInfo tracks key-value information, usually about a whole target.
from prometheus_client import Info
i = Info('my_build_version', 'Description of info')
i.info({'version': '1.2.3', 'buildhost': 'foo@bar'})Enum tracks which of a set of states something is currently in.
from prometheus_client import Enum
e = Enum('my_task_state', 'Description of enum',
states=['starting', 'running', 'stopped'])
e.state('running')All metrics can have labels, allowing grouping of related time series.
See the best practices on naming and labels.
Taking a counter as an example:
from prometheus_client import Counter
c = Counter('my_requests_total', 'HTTP Failures', ['method', 'endpoint'])
c.labels('get', '/').inc()
c.labels('post', '/submit').inc()Labels can also be passed as keyword-arguments:
from prometheus_client import Counter
c = Counter('my_requests_total', 'HTTP Failures', ['method', 'endpoint'])
c.labels(method='get', endpoint='/').inc()
c.labels(method='post', endpoint='/submit').inc()The Python client automatically exports metrics about process CPU usage, RAM,
file descriptors and start time. These all have the prefix process, and
are only currently available on Linux.
The namespace and pid constructor arguments allows for exporting metrics about other processes, for example:
ProcessCollector(namespace='mydaemon', pid=lambda: open('/var/run/daemon.pid').read())
The client also automatically exports some metadata about Python. If using Jython,
metadata about the JVM in use is also included. This information is available as
labels on the python_info metric. The value of the metric is 1, since it is the
labels that carry information.
There are several options for exporting metrics.
Metrics are usually exposed over HTTP, to be read by the Prometheus server.
The easiest way to do this is via start_http_server, which will start a HTTP
server in a daemon thread on the given port:
from prometheus_client import start_http_server
start_http_server(8000)Visit http://localhost:8000/ to view the metrics.
To add Prometheus exposition to an existing HTTP server, see the MetricsHandler class
which provides a BaseHTTPRequestHandler. It also serves as a simple example of how
to write a custom endpoint.
To use prometheus with twisted, there is MetricsResource which exposes metrics as a twisted resource.
from prometheus_client.twisted import MetricsResource
from twisted.web.server import Site
from twisted.web.resource import Resource
from twisted.internet import reactor
root = Resource()
root.putChild(b'metrics', MetricsResource())
factory = Site(root)
reactor.listenTCP(8000, factory)
reactor.run()To use Prometheus with WSGI, there is
make_wsgi_app which creates a WSGI application.
from prometheus_client import make_wsgi_app
from wsgiref.simple_server import make_server
app = make_wsgi_app()
httpd = make_server('', 8000, app)
httpd.serve_forever()Such an application can be useful when integrating Prometheus metrics with WSGI apps.
The method start_wsgi_server can be used to serve the metrics through the
WSGI reference implementation in a new thread.
from prometheus_client import start_wsgi_server
start_wsgi_server(8000)To use Prometheus with Flask we need to serve metrics through a Prometheus WSGI application. This can be achieved using Flask's application dispatching. Below is a working example.
Save the snippet below in a myapp.py file
from flask import Flask
from werkzeug.wsgi import DispatcherMiddleware
from prometheus_client import make_wsgi_app
# Create my app
app = Flask(__name__)
# Add prometheus wsgi middleware to route /metrics requests
app_dispatch = DispatcherMiddleware(app, {
'/metrics': make_wsgi_app()
})Run the example web application like this
# Install uwsgi if you do not have it
pip install uwsgi
uwsgi --http 127.0.0.1:8000 --wsgi-file myapp.py --callable app_dispatchVisit http://localhost:8000/metrics to see the metrics
The textfile collector allows machine-level statistics to be exported out via the Node exporter.
This is useful for monitoring cronjobs, or for writing cronjobs to expose metrics about a machine system that the Node exporter does not support or would not make sense to perform at every scrape (for example, anything involving subprocesses).
from prometheus_client import CollectorRegistry, Gauge, write_to_textfile
registry = CollectorRegistry()
g = Gauge('raid_status', '1 if raid array is okay', registry=registry)
g.set(1)
write_to_textfile('/configured/textfile/path/raid.prom', registry)A separate registry is used, as the default registry may contain other metrics such as those from the Process Collector.
The Pushgateway allows ephemeral and batch jobs to expose their metrics to Prometheus.
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
registry = CollectorRegistry()
g = Gauge('job_last_success_unixtime', 'Last time a batch job successfully finished', registry=registry)
g.set_to_current_time()
push_to_gateway('localhost:9091', job='batchA', registry=registry)A separate registry is used, as the default registry may contain other metrics such as those from the Process Collector.
Pushgateway functions take a grouping key. push_to_gateway replaces metrics
with the same grouping key, pushadd_to_gateway only replaces metrics with the
same name and grouping key and delete_from_gateway deletes metrics with the
given job and grouping key. See the
Pushgateway documentation
for more information.
instance_ip_grouping_key returns a grouping key with the instance label set
to the host's IP address.
If the push gateway you are connecting to is protected with HTTP Basic Auth, you can use a special handler to set the Authorization header.
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
from prometheus_client.exposition import basic_auth_handler
def my_auth_handler(url, method, timeout, headers, data):
username = 'foobar'
password = 'secret123'
return basic_auth_handler(url, method, timeout, headers, data, username, password)
registry = CollectorRegistry()
g = Gauge('job_last_success_unixtime', 'Last time a batch job successfully finished', registry=registry)
g.set_to_current_time()
push_to_gateway('localhost:9091', job='batchA', registry=registry, handler=my_auth_handler)It is also possible to expose metrics to systems other than Prometheus. This allows you to take advantage of Prometheus instrumentation even if you are not quite ready to fully transition to Prometheus yet.
Metrics are pushed over TCP in the Graphite plaintext format.
from prometheus_client.bridge.graphite import GraphiteBridge
gb = GraphiteBridge(('graphite.your.org', 2003))
# Push once.
gb.push()
# Push every 10 seconds in a daemon thread.
gb.start(10.0)Sometimes it is not possible to directly instrument code, as it is not in your control. This requires you to proxy metrics from other systems.
To do so you need to create a custom collector, for example:
from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily, REGISTRY
class CustomCollector(object):
def collect(self):
yield GaugeMetricFamily('my_gauge', 'Help text', value=7)
c = CounterMetricFamily('my_counter_total', 'Help text', labels=['foo'])
c.add_metric(['bar'], 1.7)
c.add_metric(['baz'], 3.8)
yield c
REGISTRY.register(CustomCollector())SummaryMetricFamily and HistogramMetricFamily work similarly.
A collector may implement a describe method which returns metrics in the same
format as collect (though you don't have to include the samples). This is
used to predetermine the names of time series a CollectorRegistry exposes and
thus to detect collisions and duplicate registrations.
Usually custom collectors do not have to implement describe. If describe is
not implemented and the CollectorRegistry was created with auto_desribe=True
(which is the case for the default registry) then collect will be called at
registration time instead of describe. If this could cause problems, either
implement a proper describe, or if that's not practical have describe
return an empty list.
Prometheus client libaries presume a threaded model, where metrics are shared across workers. This doesn't work so well for languages such as Python where it's common to have processes rather than threads to handle large workloads.
To handle this the client library can be put in multiprocess mode. This comes with a number of limitations:
- Registries can not be used as normal, all instantiated metrics are exported
- Custom collectors do not work (e.g. cpu and memory metrics)
- Info and Enum metrics do not work
- The pushgateway cannot be used
- Gauges cannot use the
pidlabel
There's several steps to getting this working:
One: Deployment
The prometheus_multiproc_dir environment variable must be set to a directory
that the client library can use to share metric databases between processes. In production it is recommended
to use a tmpfs volume (e.g. /tmp/prometheus) for this directory so that the exporter doesn't interfere
disk IO.
Application workers write to metric databases in this directory. The exporter process reads from it and merges dead application worker metric databases.
Add the following to your Gunicorn config file:
from prometheus_client import multiprocess_exporter
def on_starting(server):
multiprocess.start_archiver_thread()Add the Prometheus Exporter WSGI handler to your existing WSGI handler
from prometheus_client import multiprocess_exporter
....
def app(environ, start_response):
if environ.get("PATH_INFO") == "/metrics":
return multiprocess_exporter.app(environ, start_response)
else:
return ...In the same filesystem as your other Python application, start an exporter sidecar
#!/bin/bash
export prometheus_multiproc_dir=/tmp/prometheus # some dir
# Start the sidecar exporter:
(
set -eu
mkdir -p ${prometheus_multiproc_dir}
exec gunicorn \
--config python:prometheus_client.prometheus_exporter \
--preload \
--workers 1 \
--threads 10 \
--bind 0.0.0.0:9500 \
prometheus_client.prometheus_exporter:app
) &
# Start the application:
celery ...
This will export the metrics at http://127.0.0.1:9500
Only one exporter process should run per filesystem, prometheus_multiproc_dir.
Two: Inside the application
from prometheus_client import multiprocess
from prometheus_client import generate_latest, CollectorRegistry, CONTENT_TYPE_LATEST, Gauge
# Example gauge.
IN_PROGRESS = Gauge("inprogress_requests", "help", multiprocess_mode='livesum')
# Expose metrics.
@IN_PROGRESS.track_inprogress()
def app(environ, start_response):
registry = CollectorRegistry()
multiprocess.MultiProcessCollector(registry)
data = generate_latest(registry)
status = '200 OK'
response_headers = [
('Content-type', CONTENT_TYPE_LATEST),
('Content-Length', str(len(data)))
]
start_response(status, response_headers)
return iter([data])Three: Instrumentation
Counters, Summarys and Histograms work as normal.
Gauges have several modes they can run in, which can be selected with the
multiprocess_mode parameter.
- 'all': Default. Return a timeseries per process alive or dead.
- 'latest': Default. Return the most recent gauge update value.
- 'liveall': Return a timeseries per process that is still alive.
- 'livesum': Return a single timeseries that is the sum of the values of alive processes.
- 'max': Return a single timeseries that is the maximum of the values of all processes, alive or dead.
- 'min': Return a single timeseries that is the minimum of the values of all processes, alive or dead.
The Python client supports parsing the Prometheus text format. This is intended for advanced use cases where you have servers exposing Prometheus metrics and need to get them into some other system.
from prometheus_client.parser import text_string_to_metric_families
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))