authentik.root.monitoring

Metrics view

 1"""Metrics view"""
 2
 3from django.db import connections
 4from django.db.utils import OperationalError
 5from django.dispatch import Signal
 6from django.http import HttpRequest, HttpResponse
 7from django.views import View
 8from django_prometheus.exports import ExportToDjangoView
 9
10monitoring_set = Signal()
11
12
13class MetricsView(View):
14    """Wrapper around ExportToDjangoView with authentication, accessed by the authentik router"""
15
16    def get(self, request: HttpRequest) -> HttpResponse:
17        """Check for HTTP-Basic auth"""
18        monitoring_set.send_robust(self)
19        return ExportToDjangoView(request)
20
21
22class LiveView(View):
23    """View for liveness probe, always returns Http 200"""
24
25    def dispatch(self, request: HttpRequest) -> HttpResponse:
26        return HttpResponse(status=200)
27
28
29class ReadyView(View):
30    """View for readiness probe, always returns Http 200, unless sql is down"""
31
32    def check_db(self):
33        for db_conn in connections.all():
34            # Force connection reload
35            db_conn.connect()
36            _ = db_conn.cursor()
37
38    def dispatch(self, request: HttpRequest) -> HttpResponse:
39        try:
40            self.check_db()
41        except OperationalError:  # pragma: no cover
42            return HttpResponse(status=503)
43        return HttpResponse(status=200)
monitoring_set = <django.dispatch.dispatcher.Signal object>
class MetricsView(django.views.generic.base.View): 
14class MetricsView(View):
15    """Wrapper around ExportToDjangoView with authentication, accessed by the authentik router"""
16
17    def get(self, request: HttpRequest) -> HttpResponse:
18        """Check for HTTP-Basic auth"""
19        monitoring_set.send_robust(self)
20        return ExportToDjangoView(request)

Wrapper around ExportToDjangoView with authentication, accessed by the authentik router

def get( self, request: django.http.request.HttpRequest) -> django.http.response.HttpResponse: 
17    def get(self, request: HttpRequest) -> HttpResponse:
18        """Check for HTTP-Basic auth"""
19        monitoring_set.send_robust(self)
20        return ExportToDjangoView(request)

Check for HTTP-Basic auth

class LiveView(django.views.generic.base.View): 
23class LiveView(View):
24    """View for liveness probe, always returns Http 200"""
25
26    def dispatch(self, request: HttpRequest) -> HttpResponse:
27        return HttpResponse(status=200)

View for liveness probe, always returns Http 200

def dispatch( self, request: django.http.request.HttpRequest) -> django.http.response.HttpResponse: 
26    def dispatch(self, request: HttpRequest) -> HttpResponse:
27        return HttpResponse(status=200)
class ReadyView(django.views.generic.base.View): 
30class ReadyView(View):
31    """View for readiness probe, always returns Http 200, unless sql is down"""
32
33    def check_db(self):
34        for db_conn in connections.all():
35            # Force connection reload
36            db_conn.connect()
37            _ = db_conn.cursor()
38
39    def dispatch(self, request: HttpRequest) -> HttpResponse:
40        try:
41            self.check_db()
42        except OperationalError:  # pragma: no cover
43            return HttpResponse(status=503)
44        return HttpResponse(status=200)

View for readiness probe, always returns Http 200, unless sql is down

def check_db(self): 
33    def check_db(self):
34        for db_conn in connections.all():
35            # Force connection reload
36            db_conn.connect()
37            _ = db_conn.cursor()
def dispatch( self, request: django.http.request.HttpRequest) -> django.http.response.HttpResponse: 
39    def dispatch(self, request: HttpRequest) -> HttpResponse:
40        try:
41            self.check_db()
42        except OperationalError:  # pragma: no cover
43            return HttpResponse(status=503)
44        return HttpResponse(status=200)