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