Twisted Support#

The Twisted-related APIs can be found within the prometheus_async.tx package.

Decorator Wrappers#

prometheus_async.tx.time(metric: Observer) Callable[[Callable[P, D]], Callable[P, D]]#
prometheus_async.tx.time(metric: Observer, deferred: D) D

Call metric.observe(time) with runtime in seconds.

Can be used as a decorator as well as on Deferreds.

Works with both sync and async results.

Returns

function or Deferred.

The fact it’s accepting Deferreds is useful in conjunction with twisted.web views that don’t allow to return a Deferred:

from prometheus_client import Histogram
from prometheus_async.tx import time
from twisted.internet.task import deferLater
from twisted.web.resource import Resource
from twisted.web.server import NOT_DONE_YET
from twisted.internet import reactor

REQ_TIME = Histogram("req_time_seconds", "time spent in requests")

class DelayedResource(Resource):
   def _delayedRender(self, request):
      request.write("<html><body>Sorry to keep you waiting.</body></html>")
      request.finish()

   def render_GET(self, request):
      d = deferLater(reactor, 5, lambda: request)
      time(REQ_TIME, d.addCallback(self._delayedRender))
      return NOT_DONE_YET
prometheus_async.tx.count_exceptions(metric: Incrementer, *, exc: type[BaseException] = <class 'BaseException'>) Callable[P, C]#
prometheus_async.tx.count_exceptions(metric: Incrementer, deferred: D, *, exc: type[BaseException] = <class 'BaseException'>) D

Call metric.inc() whenever exc is caught.

Can be used as a decorator or on a Deferred.

Returns

function (if decorator) or Deferred.

prometheus_async.tx.track_inprogress(metric: Gauge) Callable[P, C]#
prometheus_async.tx.track_inprogress(metric: Gauge, deferred: D) D

Call metrics.inc() on entry and metric.dec() on exit.

Can be used as a decorator or on a Deferred.

Returns

function (if decorator) or Deferred.

Metric Exposure#

prometheus_client, the underlying Prometheus client library, exposes a twisted.web.resource.Resource – namely prometheus_client.twisted.MetricsResource – that makes it extremely easy to expose your metrics.

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()

As a slightly more in-depth example, the following exposes the application’s metrics under /metrics and sets up a prometheus_client.Counter for inbound HTTP requests. It also uses Klein to set up the routes instead of relying directly on twisted.web for routing.

from prometheus_client.twisted import MetricsResource
from twisted.web.server import Site
from twisted.internet import reactor

from klein import Klein

from prometheus_client import Counter


INBOUND_REQUESTS = Counter(
   "inbound_requests_total",
   "Counter (int) of inbound http requests",
   ["endpoint", "method"]
)

app = Klein()

@app.route("/metrics")
def metrics(request):
    INBOUND_REQUESTS.labels("/metrics", "GET").inc()
    return MetricsResource()


factory = Site(app.resource())
reactor.listenTCP(8000, factory)
reactor.run()