[flask]集成Prometheus

前言

在服务监控中，大抵可分为日志监控和指标监控。日志监控一样平常由类似ELK如许的日志系统去收集分析，而指标监控一样平常是由Prometheus收集服务的一些可量化的指标数据，好比服务响应码、响应时间。
安装sdk

1. python -m pip install prometheus-client

复制代码

示例

示例1

使用 prometheus_client自带的make_wsgi_app直接绑定到flask实例上，以下client使用的registry是自带默认的REGISTRY。

1. from flask import Flask
2. from werkzeug.middleware.dispatcher import DispatcherMiddleware
3. from prometheus_client import make_wsgi_app, Info
5. i = Info("my_build_version", "Description of info")
6. i.info({"version": "1.2.3", "buildhost": "foo@bar"})
9. app = Flask(__name__)
11. app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/metrics": make_wsgi_app()})
14. @app.get("/")
15. def hello():
16.     return "Hello World!"
19. if __name__ == "__main__":
20.     app.run(debug=False)

复制代码

访问http://127.0.0.1:5000/metrics可以看到类似以下的输出，其中包含了自定义的metric my_build_version_info

1. # HELP python_gc_objects_collected_total Objects collected during gc
2. # TYPE python_gc_objects_collected_total counter
3. python_gc_objects_collected_total{generation="0"} 527.0
4. python_gc_objects_collected_total{generation="1"} 124.0
5. python_gc_objects_collected_total{generation="2"} 0.0
6. # HELP python_gc_objects_uncollectable_total Uncollectable objects found during GC
7. # TYPE python_gc_objects_uncollectable_total counter
8. python_gc_objects_uncollectable_total{generation="0"} 0.0
9. python_gc_objects_uncollectable_total{generation="1"} 0.0
10. python_gc_objects_uncollectable_total{generation="2"} 0.0
11. # HELP python_gc_collections_total Number of times this generation was collected
12. # TYPE python_gc_collections_total counter
13. python_gc_collections_total{generation="0"} 112.0
14. python_gc_collections_total{generation="1"} 10.0
15. python_gc_collections_total{generation="2"} 0.0
16. # HELP python_info Python platform information
17. # TYPE python_info gauge
18. python_info{implementation="CPython",major="3",minor="11",patchlevel="2",version="3.11.2"} 1.0
19. # HELP process_virtual_memory_bytes Virtual memory size in bytes.
20. # TYPE process_virtual_memory_bytes gauge
21. process_virtual_memory_bytes 5.11332352e+08
22. # HELP process_resident_memory_bytes Resident memory size in bytes.
23. # TYPE process_resident_memory_bytes gauge
24. process_resident_memory_bytes 4.4941312e+07
25. # HELP process_start_time_seconds Start time of the process since unix epoch in seconds.
26. # TYPE process_start_time_seconds gauge
27. process_start_time_seconds 1.74637360939e+09
28. # HELP process_cpu_seconds_total Total user and system CPU time spent in seconds.
29. # TYPE process_cpu_seconds_total counter
30. process_cpu_seconds_total 0.25
31. # HELP process_open_fds Number of open file descriptors.
32. # TYPE process_open_fds gauge
33. process_open_fds 7.0
34. # HELP process_max_fds Maximum number of open file descriptors.
35. # TYPE process_max_fds gauge
36. process_max_fds 1.048576e+06
37. # HELP my_build_version_info Description of info
38. # TYPE my_build_version_info gauge
39. my_build_version_info{buildhost="foo@bar",version="1.2.3"} 1.0

复制代码

如果不想要自带输出的这些gc信息，或者想要手动注册，也可以自定义registry，

1. from flask import Flask
2. from werkzeug.middleware.dispatcher import DispatcherMiddleware
3. from prometheus_client import make_wsgi_app, Info, CollectorRegistry, PROCESS_COLLECTOR, GC_COLLECTOR, PLATFORM_COLLECTOR
5. registry = CollectorRegistry(auto_describe=True)
6. registry.register(PROCESS_COLLECTOR)
7. registry.register(GC_COLLECTOR)
8. registry.register(PLATFORM_COLLECTOR)
10. i = Info("my_build_version", "Description of info", registry=registry)
11. i.info({"version": "1.2.3", "buildhost": "foo@bar"})
14. app = Flask(__name__)
16. app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/metrics": make_wsgi_app(registry=registry)})
19. @app.get("/")
20. def hello():
21.     return "Hello World!"
24. if __name__ == "__main__":
25.     app.run(debug=False)

复制代码

示例2

1. .
2. ├── flaskapp.py
3. ├── pkg
4. │   ├── __init__.py
5. │   └── metrics
6. │       ├── __init__.py
7. │       └── metrics.py
8. └── router
9.     ├── __init__.py
10.     └── metrics.py

复制代码

• flaskapp.py
  

1. from flask import Flask, request, Response
2. from pkg.metrics import request_counter
3. from router import metrics
4. from random import randint
5. import json
7. app = Flask(__name__)
10. @app.get("/")
11. def hello():
12.     status_list = [101, 200, 401, 403, 404, 499, 500, 503, 504]
13.     request_counter.labels(
14.         method=request.method,
15.         path=request.path,
16.         status=status_list[randint(0, len(status_list) - 1)],
17.     ).inc()
18.     return {"message": "Hello World!"}
20. @app.post("/webhook")
21. def jsontest():
22.     # 获取消息头
23.     print(request.headers)
25.         # 接收数据并转成字典
26.     data = request.get_data()
27.     # data = json.loads(data)
28.     print(data)
30.     # 响应json格式数据
31.     return Response(status=200)
33. app.register_blueprint(metrics.bp)
35. if __name__ == "__main__":
36.     app.run(host="0.0.0.0", port=8000)

复制代码

• pkg/metrics/__init__.py
  

1. from .metrics import registry, request_counter
3. __all__ = ["registry", "request_counter"]

复制代码

• pkg/metrics/metrics.py
  

1. from prometheus_client import Counter, CollectorRegistry, Info
2. import socket
3. from functools import lru_cache
4. import sys
6. registry = CollectorRegistry(auto_describe=True)
8. request_counter = Counter(
9.     "http_requests_total",
10.     "Total HTTP Requests",
11.     ["method", "path", "status"],
12.     registry=registry,
13. )
15. @lru_cache(maxsize=128)
16. def get_selfip():
17.     try:
18.         s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
19.         s.connect(('8.8.8.8', 53))
20.         return s.getsockname()[0]
21.     except Exception:
22.         return "127.0.0.1"
24. @lru_cache(maxsize=None)
25. def get_pyversion():
26.     return sys.version.split(" ")[0]
28. app_info = Info(
29.     "app_info",
30.     "Application info",
31.     registry=registry,
32. )
34. app_info.info({
35.     "version": "1.0.0",
36.     "host": get_selfip(),
37.     "python_version": get_pyversion(),
38. })

复制代码

• router/__init__.py内容为空
  
• router/metrics.py
  

1. from flask import Blueprint, make_response
2. from pkg.metrics import registry
3. from prometheus_client import generate_latest
5. bp = Blueprint("metrics", __name__, url_prefix="/metrics")
7. @bp.get("/prometheus")
8. def get_metrics():
9.     """
10.     获取 Prometheus 的指标数据
11.     """
12.     resp = make_response(generate_latest(registry))
13.     resp.headers["Content-Type"] = "text/plain"
14.     return resp

复制代码

示例3-在中间件中采集http请求

1. from flask import Flask
2. from werkzeug.middleware.dispatcher import DispatcherMiddleware
3. from prometheus_client import (
4.     Counter,
5.     Histogram,
6.     make_wsgi_app,
7.     Info,
8. )
9. import time
10. from random import random, randint
13. class Metrics:
14.     def __init__(self):
15.         self.app_info = Info("python_service_basic_info", "Description of info")
16.         self.request_counter = Counter(
17.             "http_request_total", "Total HTTP requests", ["method", "path", "status"]
18.         )
19.         self.response_time = Histogram(
20.             "http_response_time",
21.             "HTTP response time",
22.             ["method", "path"],
23.         )
24.         self._initialize()
26.     def _initialize(self):
27.         self.app_info.info(
28.             {
29.                 "version": self._get_app_info_version(),
30.                 "name": "myapp",
31.             }
32.         )
34.     def _get_app_info_version(self) -> str:
35.         return "0.1.0"
38. collector = Metrics()
41. class MetricsMiddleware:
42.     def __init__(self, app):
43.         self.app = app
44.         self.white_list = frozenset(
45.             [
46.                 "/metrics",
47.                 "/health",
48.             ]
49.         )
51.     def __call__(self, environ, start_response):
52.         method = environ.get("REQUEST_METHOD", "NaN")
53.         path = environ.get("PATH_INFO", "NaN")
54.         resp_status_code = None
56.         def catching_start_response(status, headers, exc_info=None):
57.             nonlocal resp_status_code
58.             resp_status_code = status.split(" ")[0]
59.             return start_response(status, headers, exc_info)
61.         start_time = time.time()
62.         response = self.app(environ, catching_start_response)
63.         response_time = round(time.time() - start_time, 4)
65.         if path not in self.white_list:
66.             collector.request_counter.labels(method, path, resp_status_code).inc()
67.             collector.response_time.labels(method, path).observe(response_time)
69.         return response
72. app = Flask(__name__)
73. app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/metrics": make_wsgi_app()})
74. app.wsgi_app = MetricsMiddleware(app.wsgi_app)
77. @app.get("/a1")
78. def a1():
79.     return "a1"
82. @app.get("/a2")
83. def a2():
84.     time.sleep(random())
85.     return "a2"
88. @app.get("/a3")
89. def a3():
90.     time.sleep(randint(1, 3))
91.     return "a3"
94. @app.get("/a4")
95. def a4():
96.     time.sleep(randint(1, 3))
97.     raise Exception("a4")
100. if __name__ == "__main__":
101.     app.run(host="127.0.0.1", port=5000, debug=False)

复制代码

prometheus配置

1. scrape_configs:
2.   # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config.
3.   - job_name: "python"
4.     metrics_path: '/metrics/prometheus'
5.     file_sd_configs:
6.     - files: ['sd_configs/python/*.yaml']
7.       refresh_interval:  10s

复制代码

python/target.yaml

1. - targets: ['192.168.1.112:8000']
2.   labels:
3.     instance: 192.168.1.112

复制代码

参考

• prometheus client-python 官方文档
  
• Prometheus监控的4个黄金指标及示例
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。