郑文峰的博客
zhengwenfeng
2024-09-17
目录

django-prometheus使用及源码分析

# 简介

在django服务运行过程中，希望可以对其获取promethues指标进行监控，这样可以实时知道其运行状态，当它运行异常时可以及时进行告警，并且帮助我们可以对其针对性进行优化。比如请求量过大是否要进行限流或者扩容，再或者发现接口过慢，可能是数据库访问太慢，出现了慢sql，需要及时进行优化等等。

而本文主要是介绍使用django-prometheus (opens new window)来对django服务添加对prometheus指标的支持，它已经内置了部分的指标采集，包括请求、数据库和缓存等方面的指标。除了使用方法外，也会对其源码进行分析，看它是如何实现的。

本文中使用的例子已经上传到github中，可以在django_demo (opens new window)上查看，搭配本文章学习。

‍

# 获取prometheus指标

# 新增接口获取指标

在url.py中新增下面的路由

path('', include('django_prometheus.urls')),
1

然后运行服务，调用/metrics接口，即可获取到默认的prometheus指标信息。

...
python_gc_objects_collected_total{generation="0"} 667.0
python_gc_objects_collected_total{generation="1"} 405.0
python_gc_objects_collected_total{generation="2"} 19.0
...
1
2
3
4
5

我们可以看下引用的​django_prometheus.urls源码，它包含了一个metrics的路由，调用的是​exports.ExportToDjangoView方法

urlpatterns = [path("metrics", exports.ExportToDjangoView, name="prometheus-django-metrics")]
1

再看看下​ExportToDjangoView方法，这里有一个分支，如果配置了环境变量PROMETHEUS_MULTIPROC_DIR或者​prometheus_multiproc_dir则会走多进程收集指标逻辑，否则单进程会则从全局变量REGISTRY​中获取所有的指标，最后返回响应。

def ExportToDjangoView(request):
    if "PROMETHEUS_MULTIPROC_DIR" in os.environ or "prometheus_multiproc_dir" in os.environ:
        registry = prometheus_client.CollectorRegistry()
        multiprocess.MultiProcessCollector(registry)
    else:
        registry = prometheus_client.REGISTRY
    metrics_page = prometheus_client.generate_latest(registry)
    return HttpResponse(metrics_page, content_type=prometheus_client.CONTENT_TYPE_LATEST)
1
2
3
4
5
6
7
8

这里注意多进程与单进程收集指标的方式是不一样的，多进程是从各个进程的文件读取，而单进程是从全局变量中读取。

‍

# 在专用线程中获取指标

上面的方法是在django服务中获取指标，但如果业务bug可能会导致监控受到影响，出现无法获取到指标的情况，这样就无法提供定位问题的帮助。

所以提供了一种方式在单独的线程中来获取指标，达到解耦的目的，保证即使业务异常也不会影响指标的获取。

这种方式默认是关闭的，需要在setting.py中添加以下变量启用：

PROMETHEUS_METRICS_EXPORT_PORT = 8001
PROMETHEUS_METRICS_EXPORT_ADDRESS = '0.0.0.0'  # all addresses
1
2

然后需要在 INSTALLED_APPS ​中添加 django_prometheus ​，这时因为该线程是在 DjangoPrometheusConfig.ready ​-> SetupPrometheusExportsFromConfig ​-> SetupPrometheusEndpointOnPort ​中被调用的，需要引用该app才会被执行。

INSTALLED_APPS = [
    ...
    'django_prometheus',
    'demo',
]
1
2
3
4
5

再来看 SetupPrometheusEndpointOnPort ​方法可以看到是调用 prometheus_client.start_http_server ​方法来开启线程，并暴露指定的 PROMETHEUS_METRICS_EXPORT_PORT ​端口和允许访问的 PROMETHEUS_METRICS_EXPORT_ADDRESS ​地址。

def SetupPrometheusEndpointOnPort(port, addr=""):
    assert os.environ.get("RUN_MAIN") != "true", (
        "The thread-based exporter can't be safely used when django's "
        "autoreloader is active. Use the URL exporter, or start django "
        "with --noreload. See documentation/exports.md."
    )
    prometheus_client.start_http_server(port, addr=addr)
1
2
3
4
5
6
7

注意这里有一个断言，该种方式不支持热加载启动。

最后运行服务，通过访问上面的配置的8001端口即可获取该服务的默认promethues指标了。

# 请求指标

通过上面的步骤，已经知道了如何配置获取指标信息，现在需要知道如何获取请求质量的指标信息。

而 django_prometheus 已经给我们默认提供了请求相关的指标信息，部分如下：

指标 说明 阶段
django_http_requests_latency_including_middlewares_seconds 包含middleware的请求时间直方图 在开始的middleware process_request
django_http_requests_before_middlewares_total 请求数 在开始的middleware process_request
django_http_responses_before_middlewares_total 响应数 在开始的middleware process_response
django_http_requests_latency_seconds_by_view_method view层请求时间的直方图 在最后的middlewar process_request
django_http_requests_total_by_method view层带有请求方法的请求数 在最后的middlewar process_request
django_http_requests_total_by_view_transport_method view层的请求数 在最后的middlewar process_view
responses_by_status_view_method view层的响应数 在最后的middlewar process_response

使用方法

在 settings.py 中的 MIDDLEWARE ​中添加两个中间件，配置如下：

MIDDLEWARE = [
    'django_prometheus.middleware.PrometheusBeforeMiddleware',
  ...
    'django_prometheus.middleware.PrometheusAfterMiddleware',
]
1
2
3
4
5

注意这里的顺序很重要，一定要是放在MIDDLEWARE的第一个和最后一个。

在 view.py ​中新增一个接口 myview

def my_view(request):
    return HttpResponse("hello")
1
2

在 urls.py 中新增一条路由

path('myview/', my_view)
1

最后你请求接口一次该接口，再获取指标，你可以得到部分请求指标的变化。

django_http_requests_latency_seconds_by_view_method_bucket{le="0.01",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.025",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.05",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.075",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.1",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.25",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.5",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="0.75",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="1.0",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="2.5",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="5.0",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="7.5",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="10.0",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="25.0",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="50.0",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="75.0",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_bucket{le="+Inf",method="GET",view="demo.views.my_view"} 1.0
django_http_requests_latency_seconds_by_view_method_count{method="GET",view="demo.views.my_view"} 1.0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

实现原理

请求进入 view 层之前会先按照顺序经过 middleware，然后再到view进行执行，最后响应的时候再按照逆序经过 middleware，如下图所示：

1726558982722image-20240917134902-uu4hpp3.png

而 middleware 中会有一系列的钩子函数可以对请求做一些预处理工作，比如认证等，而我们的请求指标就是在 middleware 层中实现的。

先看 PrometheusBeforeMiddleware，实现了 process_request 和 process_response 方法，这两个方法就是在 request 进来时和 response 返回时分别会调用的方法。

class PrometheusBeforeMiddleware(MiddlewareMixin):
    metrics_cls = Metrics

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.metrics = self.metrics_cls.get_instance()

    def process_request(self, request):
        self.metrics.requests_total.inc()
        request.prometheus_before_middleware_event = Time()

    def process_response(self, request, response):
        self.metrics.responses_total.inc()
        if hasattr(request, "prometheus_before_middleware_event"):
            self.metrics.requests_latency_before.observe(TimeSince(request.prometheus_before_middleware_event))
        else:
            self.metrics.requests_unknown_latency_before.inc()
        return response
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

如下图所示：

1726559026319image-20240917135711-b83sfjz.png

在 process_request 中对 requests_total 进行递增，在 process_response 中对 responses_total 也进行递增，这两个就是进入服务的请求总数和响应总数的指标。

再看 process_request 中记录了当前的时间 prometheus_before_middleware_event，然后在 process_response 中将当前时间与上一个时间进行相减，也就得到了从请求进来到响应的整个时间，赋值给了 requests_latency_before 指标中。

而除了配置 PrometheusBeforeMiddleware 外，还配置了 PrometheusAfterMiddleware，但基本原理都是一样的，都是通过中间件在不同阶段的钩子方法来进行指标的统计，只是统计的指标不一样而已。

‍

# postgres指标

接下来就是讲关于数据库的指标，该库支持mysql、sqlite和postgres等数据库的支持，但这里主要是对postgres介绍，其他的使用方法也是类似。

# 对insert、update、delete操作计数

使用方法

在 models.py 中创建 class User，并继承 ExportModelOperationsMixin，这是一个常见的python设计模式，可以给已有的类额外的添加功能。

from django_prometheus.models import ExportModelOperationsMixin

class User(ExportModelOperationsMixin('User'), models.Model):
    class Meta:
        db_table = 't_user'

    id = models.AutoField(primary_key=True)
    name = models.CharField()
    age = models.IntegerField()
    sex = models.CharField()
1
2
3
4
5
6
7
8
9
10

然后建表

python manager.py makemigrations
python manager.py migrate
1
2

在view.py新增接口my_view2

def my_view2(request):
    User(name="zhangsan", age=12, sex="male").save()
    return HttpResponse("hello")
1
2
3
4

在url.py新增路由

path('myview2/', my_view2)
1

然后运行服务，先调用/myview2接口，再获取指标就可以看到对User的新增计数

django_model_inserts_total{model="User"} 1.0
1

实现原理

我们看看 ExportModelOperationsMixin 类，可以看到里面创建了一个 class Mixin 它重写了 _do_insert、 _do_update 、 delete 三个方法，而这三个方法是 models.Model 的方法，是在对 model 进行插入、更新和删除时执行的三个方法，而这里重写是在执行这些操作前，使用指标变量进行递增，这样就记录了次数。

def ExportModelOperationsMixin(model_name):
    model_inserts.labels(model_name)
    model_updates.labels(model_name)
    model_deletes.labels(model_name)

    class Mixin:
        def _do_insert(self, *args, **kwargs):
            model_inserts.labels(model_name).inc()
            return super()._do_insert(*args, **kwargs)

        def _do_update(self, *args, **kwargs):
            model_updates.labels(model_name).inc()
            return super()._do_update(*args, **kwargs)

        def delete(self, *args, **kwargs):
            model_deletes.labels(model_name).inc()
            return super().delete(*args, **kwargs)

    Mixin.__qualname__ = f"ExportModelOperationsMixin('{model_name}')"
    return Mixin
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 执行耗时

除了执行次数外，我们还想知道数据库执行的耗时。

使用方法

在 settings.py 中，将 DATABSE 中的 engine 换成 django_prometheus.db.backends.postgresql

DATABASES = {
    'default': {
        'ENGINE': 'django_prometheus.db.backends.postgresql',
    ...
    }
}
1
2
3
4
5
6
7

然后就可以再次调用接口 myview2 对 User 进行操作，再获取指标可以看到对 postgres 的耗时的一个直方图指标。

django_db_query_duration_seconds_bucket{alias="default",le="0.01",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.025",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.05",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.075",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.1",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.25",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.5",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="0.75",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="1.0",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="2.5",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="5.0",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="7.5",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="10.0",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="25.0",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="50.0",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="75.0",vendor="postgresql"} 3.0
django_db_query_duration_seconds_bucket{alias="default",le="+Inf",vendor="postgresql"} 3.0
django_db_query_duration_seconds_count{alias="default",vendor="postgresql"} 3.0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

实现原理

我们看向 django_prometheus/db/backends/postgresql/base.py 文件，可以看到 DatabaseWrapper 类，而该类重写了 get_new_connection 方法，它目的是在创建数据库连接时，将内置使用的cursor替换成自己定义的 ExportingCursorWrapper

class DatabaseWrapper(DatabaseWrapperMixin, base.DatabaseWrapper):
    def get_new_connection(self, *args, **kwargs):
        conn = super().get_new_connection(*args, **kwargs)
        conn.cursor_factory = ExportingCursorWrapper(
            conn.cursor_factory or get_postgres_cursor_class(), self.alias, self.vendor
        )
        return conn

    def create_cursor(self, name=None):
        # cursor_factory is a kwarg to connect() so restore create_cursor()'s
        # default behavior
        return base.DatabaseWrapper.create_cursor(self, name=name)
1
2
3
4
5
6
7
8
9
10
11
12
13

再看向 class ExportingCursorWrapper，里面定义了 class CursorWrapper，重写了 execute 和 executemany 两个方法，是在执行sql时会调用的方法，而在调用父类执行sql前后进行统计计数和耗时。

def ExportingCursorWrapper(cursor_class, alias, vendor):
    labels = {"alias": alias, "vendor": vendor}

    class CursorWrapper(cursor_class):
        def execute(self, *args, **kwargs):
            execute_total.labels(alias, vendor).inc()
            with query_duration_seconds.labels(**labels).time(), ExceptionCounterByType(
                errors_total, extra_labels=labels
            ):
                return super().execute(*args, **kwargs)

        def executemany(self, query, param_list, *args, **kwargs):
            execute_total.labels(alias, vendor).inc(len(param_list))
            execute_many_total.labels(alias, vendor).inc(len(param_list))
            with query_duration_seconds.labels(**labels).time(), ExceptionCounterByType(
                errors_total, extra_labels=labels
            ):
                return super().executemany(query, param_list, *args, **kwargs)

    return CursorWrapper
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

指标列表

指标 说明
django_db_query_duration_seconds 数据库所有的操作耗时，包含了增删改查
django_db_new_connections_total 数据库创建的连接数
execute_total 数据库执行总数

‍

# redis指标

使用方法

在 settings.py 中配置redis

CACHES = {
    'default': {
        'BACKEND': 'django_prometheus.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379'
    }
}
1
2
3
4
5
6

在 view.py 中新增 my_view3 接口，对redis进行操作

def my_view3(request):
    cache.set("name", "zhangsan")
    return HttpResponse(cache.get("name"))
1
2
3

在url.py中新增对应路由

path('myview3/', my_view3)
1

调用接口 /myview3 接口，然后获取指标，就可以看到redis相关的指标参数

django_cache_get_total{backend="redis"} 1.0
django_cache_get_created{backend="redis"} 1.7265568108417125e+09
django_cache_get_hits_total{backend="redis"} 1.0
django_cache_get_hits_created{backend="redis"} 1.726556810842234e+09
1
2
3
4

实现原理

我们看到 django_prometheus/cache/backends/redis.py 中的class RedisCache，该类继承了已有的缓存类 cache.RedisCache，重写了方法get，在调用父类操作方法的前后添加指标。

class RedisCache(cache.RedisCache):
    @cache.omit_exception
    def get(self, key, default=None, version=None, client=None):
        try:
            django_cache_get_total.labels(backend="redis").inc()
            cached = self.client.get(key, default=None, version=version, client=client)
        except exceptions.ConnectionInterrupted as e:
            django_cache_get_fail_total.labels(backend="redis").inc()
            if self._ignore_exceptions:
                if self._log_ignored_exceptions:
                    cache.logger.error(str(e))
                return default
            raise
        else:
            if cached is not None:
                django_cache_hits_total.labels(backend="redis").inc()
                return cached
            else:
                django_cache_misses_total.labels(backend="redis").inc()
                return default
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

‍

# 总结

django-prometheus 是在充分的理解了django本身的机制上，通过已提供的钩子，或者重写部分的方法，替换掉内置的组件的方式，来达到自己的目的，这种方法值得参考与学习。

‍

#python#django
上次更新: 2024/09/17, 16:41:35
django rest_framework 分页
Flask使用flask_socketio实现websocket

← Flask使用flask_socketio实现websocket→

最近更新
01
kube-proxy源码分析
01-18
02
kubernetes service如何通过iptables转发
01-18
03
tcp缓存引起的日志丢失
11-09
更多文章>
Theme by Vdoing | Copyright © 2022-2024 zhengwenfeng | MIT License