Python3.11监控告警开发:Prometheus集成部署教程
Python3.11监控告警开发:Prometheus集成部署教程
你是不是也遇到过这种情况?线上服务半夜突然挂了,第二天早上才发现,用户投诉已经堆满了邮箱。或者,某个API接口响应越来越慢,直到用户流失了才后知后觉。
在今天的运维和开发工作中,监控告警已经不是“锦上添花”,而是“雪中送炭”的必需品。它能让你在问题发生的第一时间就收到警报,而不是等到用户来告诉你。
今天,我们就来聊聊如何用Python 3.11,结合当下最流行的监控系统Prometheus,搭建一套属于自己的监控告警体系。整个过程就像搭积木一样简单,即使你之前没接触过监控系统,也能跟着一步步做出来。
1. 为什么选择Python 3.11和Prometheus?
在开始动手之前,我们先简单了解一下今天要用到的两个核心工具。
Python 3.11 是Python语言的一个重大版本更新,相比之前的版本,它在性能上有了显著的提升——官方数据显示平均提速25%。这意味着你的监控采集脚本能跑得更快,消耗更少的资源。对于需要7x24小时运行的监控任务来说,这一点尤其重要。
Prometheus 则是一个开源的系统监控和警报工具包,现在已经成为云原生监控领域的事实标准。它最大的特点是“拉取”模式:Prometheus服务器会定期从你的应用那里“拉取”指标数据,而不是等着应用“推送”过来。这种设计让系统更加健壮和易于管理。
把它们俩结合起来,Python 3.11负责收集你应用的各种指标(比如CPU使用率、内存占用、请求数量等),然后通过一个简单的HTTP接口暴露出来;Prometheus则定期来抓取这些数据,存储到它的时间序列数据库中。一旦某个指标超过了你设定的阈值,Prometheus就会触发告警。
听起来是不是挺简单的?接下来我们就一步步实现它。
2. 环境准备:快速搭建Python 3.11开发环境
工欲善其事,必先利其器。我们先来准备好开发环境。这里我推荐使用Miniconda,它是一个轻量级的Python环境管理工具,能让你快速创建独立的开发环境,避免软件包之间的版本冲突。
2.1 获取Miniconda-Python3.11镜像
如果你在使用CSDN星图镜像,可以直接搜索“Miniconda-Python3.11”镜像。这个镜像已经预装了Python 3.11和conda包管理器,开箱即用。
启动镜像后,你可以通过两种方式使用:
Jupyter Notebook方式:通过Web界面访问,适合交互式开发和调试。你可以在浏览器中直接编写和运行Python代码,实时看到结果。
SSH方式:通过命令行连接,适合自动化脚本和后台任务。这种方式更接近生产环境的操作方式。
我个人建议在开发阶段使用Jupyter Notebook,方便调试;在部署阶段使用SSH,便于自动化。
2.2 创建独立的Python环境
虽然镜像自带了Python 3.11,但为了项目干净,我们最好创建一个独立的环境:
# 创建一个名为monitoring的新环境,指定Python版本为3.11
conda create -n monitoring python=3.11 -y
# 激活这个环境
conda activate monitoring
创建独立环境的好处是,这个项目用到的所有包都装在这个环境里,不会影响其他项目。以后如果这个项目不用了,直接删除这个环境就行,非常干净。
2.3 安装必要的Python包
接下来安装我们需要的Python包:
pip install prometheus-client flask psutil
简单介绍一下这三个包:
prometheus-client:Prometheus官方的Python客户端库,用来暴露监控指标flask:一个轻量级的Web框架,用来创建HTTP服务psutil:一个跨平台的系统监控库,能获取CPU、内存、磁盘、网络等信息
安装完成后,我们的基础环境就准备好了。
3. 第一步:用Python暴露基础监控指标
现在让我们开始写代码。首先创建一个最简单的监控端点,暴露一些基础的系统指标。
3.1 创建监控服务器脚本
新建一个文件叫 monitor_server.py,写入以下内容:
from prometheus_client import start_http_server, Gauge, Counter
import psutil
import time
# 创建几个监控指标
# Gauge类型:表示可以任意上下波动的值,比如CPU使用率、内存使用量
cpu_usage = Gauge('system_cpu_percent', 'CPU使用百分比')
memory_usage = Gauge('system_memory_percent', '内存使用百分比')
disk_usage = Gauge('system_disk_percent', '磁盘使用百分比')
# Counter类型:只增不减的计数器,比如请求总数、错误总数
request_count = Counter('http_requests_total', 'HTTP请求总数')
def collect_system_metrics():
"""收集系统指标"""
# 获取CPU使用率(1秒内的平均值)
cpu_percent = psutil.cpu_percent(interval=1)
cpu_usage.set(cpu_percent)
# 获取内存使用率
memory = psutil.virtual_memory()
memory_usage.set(memory.percent)
# 获取磁盘使用率(默认取根目录)
disk = psutil.disk_usage('/')
disk_usage.set(disk.percent)
print(f"指标已更新 - CPU: {cpu_percent}%, 内存: {memory.percent}%, 磁盘: {disk.percent}%")
if __name__ == '__main__':
# 启动一个HTTP服务器,在8000端口暴露指标
start_http_server(8000)
print("监控服务器已启动,访问 http://localhost:8000/metrics 查看指标")
# 每5秒收集一次指标
while True:
collect_system_metrics()
time.sleep(5)
这段代码做了几件事:
- 定义了三个Gauge类型的指标(CPU、内存、磁盘使用率)和一个Counter类型的指标(请求总数)
- 创建了一个函数来收集系统指标
- 启动了一个HTTP服务器,监听8000端口
- 每5秒更新一次指标数据
3.2 运行并测试监控服务器
保存文件后,在终端运行:
python monitor_server.py
你会看到输出:“监控服务器已启动,访问 http://localhost:8000/metrics 查看指标”
现在打开浏览器,访问 http://localhost:8000/metrics,你会看到类似这样的内容:
# HELP system_cpu_percent CPU使用百分比
# TYPE system_cpu_percent gauge
system_cpu_percent 15.7
# HELP system_memory_percent 内存使用百分比
# TYPE system_memory_percent gauge
system_memory_percent 45.2
# HELP system_disk_percent 磁盘使用百分比
# TYPE system_disk_percent gauge
system_disk_percent 78.3
# HELP http_requests_total HTTP请求总数
# TYPE http_requests_total counter
http_requests_total 0
这就是Prometheus能够理解的指标格式。每个指标前面有# HELP说明这个指标是什么,# TYPE说明指标类型,然后是具体的指标名和值。
4. 第二步:监控你的Python Web应用
光是监控系统资源还不够,我们更关心的是自己的应用运行得怎么样。下面我们创建一个简单的Web应用,并给它加上监控。
4.1 创建带监控的Flask应用
新建一个文件 app_with_monitor.py:
from flask import Flask, jsonify
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
import time
import random
app = Flask(__name__)
# 定义监控指标
# 请求计数器:按方法和路径统计
REQUEST_COUNT = Counter(
'http_requests_total',
'HTTP请求总数',
['method', 'endpoint', 'status']
)
# 请求耗时直方图:统计响应时间分布
REQUEST_LATENCY = Histogram(
'http_request_duration_seconds',
'HTTP请求耗时(秒)',
['method', 'endpoint']
)
# 业务特定指标:比如订单数量
ORDERS_COUNT = Counter('orders_total', '订单总数')
ERROR_COUNT = Counter('errors_total', '错误总数')
@app.route('/')
def home():
"""首页"""
start_time = time.time()
# 模拟一些处理时间
time.sleep(random.uniform(0.1, 0.3))
# 记录请求
REQUEST_COUNT.labels(method='GET', endpoint='/', status='200').inc()
# 记录耗时
REQUEST_LATENCY.labels(method='GET', endpoint='/').observe(time.time() - start_time)
return jsonify({"status": "ok", "message": "欢迎来到监控演示应用"})
@app.route('/order')
def create_order():
"""创建订单接口"""
start_time = time.time()
try:
# 模拟业务处理
time.sleep(random.uniform(0.2, 0.5))
# 10%的概率模拟一个错误
if random.random() < 0.1:
raise Exception("模拟的订单创建错误")
# 订单创建成功
ORDERS_COUNT.inc()
# 记录请求
REQUEST_COUNT.labels(method='GET', endpoint='/order', status='200').inc()
REQUEST_LATENCY.labels(method='GET', endpoint='/order').observe(time.time() - start_time)
return jsonify({"status": "success", "order_id": random.randint(1000, 9999)})
except Exception as e:
# 记录错误
ERROR_COUNT.inc()
REQUEST_COUNT.labels(method='GET', endpoint='/order', status='500').inc()
REQUEST_LATENCY.labels(method='GET', endpoint='/order').observe(time.time() - start_time)
return jsonify({"status": "error", "message": str(e)}), 500
@app.route('/metrics')
def metrics():
"""Prometheus指标端点"""
return generate_latest(), 200, {'Content-Type': CONTENT_TYPE_LATEST}
if __name__ == '__main__':
print("应用启动,访问地址:")
print("首页: http://localhost:5000")
print("订单接口: http://localhost:5000/order")
print("监控指标: http://localhost:5000/metrics")
app.run(host='0.0.0.0', port=5000, debug=False)
这个应用有三个主要端点:
/:首页,总是返回成功/order:创建订单接口,有10%的概率模拟失败/metrics:暴露监控指标,供Prometheus抓取
4.2 理解不同类型的监控指标
在这个例子中,我们用了三种不同类型的指标:
Counter(计数器):只增不减的数字,适合统计总数。比如:
http_requests_total:HTTP请求总数orders_total:订单总数errors_total:错误总数
Gauge(仪表盘):可以任意上下波动的值。比如:
- 系统CPU使用率
- 内存使用量
- 当前在线用户数
Histogram(直方图):统计数据的分布情况。比如:
http_request_duration_seconds:请求耗时的分布- 它会统计落在不同区间(比如0-0.1秒、0.1-0.5秒等)的请求数量
Histogram特别有用,因为它能告诉你“有多少请求在1秒内响应”,而不仅仅是“平均响应时间是多少”。
4.3 运行和测试应用
运行应用:
python app_with_monitor.py
然后多访问几次这几个地址:
http://localhost:5000/http://localhost:5000/orderhttp://localhost:5000/metrics
访问/metrics端点,你会看到类似这样的指标:
# HELP http_requests_total HTTP请求总数
# TYPE http_requests_total counter
http_requests_total{endpoint="/",method="GET",status="200"} 5.0
http_requests_total{endpoint="/order",method="GET",status="200"} 3.0
http_requests_total{endpoint="/order",method="GET",status="500"} 1.0
# HELP orders_total 订单总数
# TYPE orders_total counter
orders_total 3.0
# HELP errors_total 错误总数
# TYPE errors_total counter
errors_total 1.0
# HELP http_request_duration_seconds HTTP请求耗时(秒)
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.005"} 0.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.01"} 0.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.025"} 0.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.05"} 0.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.1"} 0.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.25"} 5.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="0.5"} 5.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="1.0"} 5.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="2.5"} 5.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="5.0"} 5.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="10.0"} 5.0
http_request_duration_seconds_bucket{endpoint="/",method="GET",le="+Inf"} 5.0
http_request_duration_seconds_sum{endpoint="/",method="GET"} 1.234
http_request_duration_seconds_count{endpoint="/",method="GET"} 5.0
看到那些le="0.25"、le="0.5"了吗?这就是直方图的“桶”(bucket)。le="0.25"表示“小于等于0.25秒”,上面的数据显示有5个请求的响应时间都在0.25秒以内。
5. 第三步:部署和配置Prometheus
现在我们的Python应用已经能提供监控数据了,接下来需要部署Prometheus来收集这些数据。
5.1 下载和安装Prometheus
首先下载Prometheus。访问Prometheus官网的下载页面,找到适合你系统的版本。这里以Linux为例:
# 下载Prometheus
wget https://github.com/prometheus/prometheus/releases/download/v2.45.0/prometheus-2.45.0.linux-amd64.tar.gz
# 解压
tar xvfz prometheus-2.45.0.linux-amd64.tar.gz
# 进入目录
cd prometheus-2.45.0.linux-amd64
5.2 配置Prometheus
Prometheus的配置文件是prometheus.yml。我们需要修改它,告诉Prometheus去哪里抓取我们的Python应用指标。
用文本编辑器打开prometheus.yml,找到scrape_configs部分,添加我们的应用:
# 全局配置
global:
scrape_interval: 15s # 每15秒抓取一次数据
evaluation_interval: 15s # 每15秒评估一次告警规则
# 告警规则文件
rule_files:
# - "first_rules.yml"
# - "second_rules.yml"
# 抓取配置
scrape_configs:
# 监控Prometheus自己
- job_name: 'prometheus'
static_configs:
- targets: ['localhost:9090']
# 监控我们的Python系统指标
- job_name: 'python-system'
static_configs:
- targets: ['localhost:8000'] # 这是我们第一个监控服务器的地址
# 监控我们的Flask应用
- job_name: 'python-webapp'
static_configs:
- targets: ['localhost:5000'] # 这是我们Flask应用的地址
metrics_path: '/metrics' # 指标端点路径
scrape_interval: 10s # 对这个应用,我们10秒抓取一次
这个配置告诉Prometheus:
- 每15秒抓取一次数据(全局设置)
- 监控三个目标:Prometheus自己、系统监控服务器、Flask应用
- 对Flask应用,我们更关心它的状态,所以每10秒抓取一次
5.3 启动Prometheus
确保你的两个Python应用都在运行(一个在8000端口,一个在5000端口),然后启动Prometheus:
# 在前台启动,方便看日志
./prometheus --config.file=prometheus.yml
# 或者后台启动
nohup ./prometheus --config.file=prometheus.yml > prometheus.log 2>&1 &
启动后,访问 http://localhost:9090 就能看到Prometheus的Web界面了。
5.4 在Prometheus中查看指标
在Prometheus的Web界面中,你可以:
- 在“Graph”页面输入指标名查询,比如
system_cpu_percent - 查看当前值,或者看一段时间内的趋势图
- 在“Status” → “Targets”页面查看所有监控目标的状态
如果一切正常,你应该能看到三个目标都是“UP”状态。
6. 第四步:设置告警规则
监控数据有了,但我们需要在出问题时收到告警。这就需要配置告警规则。
6.1 创建告警规则文件
在Prometheus目录下创建一个新文件 alerts.yml:
groups:
- name: example
rules:
# 规则1:CPU使用率过高
- alert: HighCPUUsage
expr: system_cpu_percent > 80
for: 2m # 持续2分钟才触发
labels:
severity: warning
annotations:
summary: "CPU使用率过高"
description: "CPU使用率已经超过80%持续2分钟,当前值: {{ $value }}%"
# 规则2:内存使用率过高
- alert: HighMemoryUsage
expr: system_memory_percent > 85
for: 2m
labels:
severity: warning
annotations:
summary: "内存使用率过高"
description: "内存使用率已经超过85%持续2分钟,当前值: {{ $value }}%"
# 规则3:错误率过高
- alert: HighErrorRate
expr: rate(errors_total[5m]) > 0.1 # 5分钟内错误率超过10%
for: 1m
labels:
severity: critical
annotations:
summary: "错误率过高"
description: "最近5分钟错误率超过10%,当前错误率: {{ $value }}"
# 规则4:请求延迟过高
- alert: HighRequestLatency
expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 1
for: 2m
labels:
severity: warning
annotations:
summary: "请求延迟过高"
description: "95%的请求延迟超过1秒,当前P95延迟: {{ $value }}秒"
# 规则5:服务宕机
- alert: ServiceDown
expr: up == 0
for: 30s
labels:
severity: critical
annotations:
summary: "服务不可用"
description: "{{ $labels.job }} 服务已经宕机"
这些规则监控了几个关键问题:
- CPU使用率超过80%
- 内存使用率超过85%
- 错误率超过10%
- 95%的请求延迟超过1秒
- 服务完全不可用
6.2 更新Prometheus配置
修改prometheus.yml,启用告警规则:
rule_files:
- "alerts.yml" # 添加这一行
重启Prometheus使配置生效。
6.3 测试告警规则
要测试告警,我们可以模拟一些异常情况。创建一个测试脚本 test_alerts.py:
import time
import requests
import random
# Flask应用的地址
APP_URL = "http://localhost:5000"
def simulate_high_traffic():
"""模拟高流量,触发延迟告警"""
print("模拟高流量...")
for i in range(50):
try:
# 访问订单接口,有概率触发错误
response = requests.get(f"{APP_URL}/order", timeout=2)
print(f"请求 {i+1}: {response.status_code}")
except Exception as e:
print(f"请求 {i+1} 失败: {e}")
time.sleep(0.1) # 每秒10个请求
def simulate_system_load():
"""模拟系统高负载"""
print("模拟CPU高负载...")
# 创建一个计算密集型的任务
start_time = time.time()
while time.time() - start_time < 30: # 运行30秒
# 做一些计算
_ = [i * i for i in range(10000)]
if __name__ == '__main__':
print("开始模拟异常情况...")
print("1. 模拟高流量")
simulate_high_traffic()
print("\n2. 模拟系统高负载")
simulate_system_load()
print("\n模拟完成,请查看Prometheus告警页面")
运行这个脚本,它会:
- 快速发送50个请求到你的Flask应用,可能触发错误率告警和延迟告警
- 运行一个计算密集型任务30秒,可能触发CPU告警
运行脚本后,回到Prometheus的Web界面,点击“Alerts”标签页,你应该能看到触发的告警。
7. 第五步:配置告警通知(可选)
Prometheus负责检测问题并触发告警,但通常我们需要把告警发送到其他地方,比如邮件、Slack、钉钉等。这需要另一个组件:Alertmanager。
7.1 安装和配置Alertmanager
下载Alertmanager:
wget https://github.com/prometheus/alertmanager/releases/download/v0.25.0/alertmanager-0.25.0.linux-amd64.tar.gz
tar xvfz alertmanager-0.25.0.linux-amd64.tar.gz
cd alertmanager-0.25.0.linux-amd64
编辑alertmanager.yml配置文件,这里以邮件通知为例:
global:
smtp_smarthost: 'smtp.example.com:587' # 你的SMTP服务器
smtp_from: 'alerts@yourcompany.com'
smtp_auth_username: 'your-email@example.com'
smtp_auth_password: 'your-password'
route:
group_by: ['alertname']
group_wait: 10s
group_interval: 10s
repeat_interval: 1h
receiver: 'email-alerts'
receivers:
- name: 'email-alerts'
email_configs:
- to: 'admin@yourcompany.com'
send_resolved: true # 问题解决时也发送通知
7.2 更新Prometheus配置
修改prometheus.yml,告诉Prometheus把告警发送给Alertmanager:
alerting:
alertmanagers:
- static_configs:
- targets:
- localhost:9093 # Alertmanager默认端口
7.3 启动Alertmanager
./alertmanager --config.file=alertmanager.yml
现在,当Prometheus触发告警时,Alertmanager会收到告警,然后根据配置发送邮件通知。
8. 实际应用建议和最佳实践
通过上面的步骤,你已经搭建了一个完整的监控告警系统。但在实际生产环境中,还有一些需要注意的地方。
8.1 监控指标的选择
不是所有数据都值得监控。好的监控系统应该关注那些真正重要的指标。我建议从这几个方面入手:
四个黄金指标(Google SRE手册推荐):
- 延迟:服务处理请求的时间
- 流量:服务的请求量
- 错误:请求失败的比例
- 饱和度:系统资源的利用率
业务指标:
- 订单成功率
- 用户活跃度
- 关键业务流程的完成率
8.2 告警策略的设计
告警太多等于没有告警。设计告警策略时要注意:
-
分级告警:
- P0(紧急):服务完全不可用,需要立即处理
- P1(重要):核心功能受影响,2小时内处理
- P2(警告):非核心问题,24小时内处理
-
避免告警风暴:
- 设置合理的
for持续时间,避免瞬时波动触发告警 - 使用告警分组,相似告警合并发送
- 设置静默期,避免重复告警
- 设置合理的
-
告警要有可操作性:
- 告警信息要包含:什么问题、在哪里发生、如何修复
- 最好能直接链接到相关日志或仪表盘
8.3 性能考虑
监控系统本身也会消耗资源,需要注意:
- 采集频率:不是越频繁越好。通常15-30秒一次就足够了
- 指标数量:每个指标都要占用存储空间。只监控必要的指标
- 数据保留:设置合理的数据保留时间。通常30-90天就够了
- 客户端性能:
prometheus-client很轻量,但如果你有数千个指标,还是要测试一下对应用的影响
8.4 代码层面的优化
在实际项目中,你可能会这样组织监控代码:
# monitor.py - 监控工具类
from prometheus_client import Counter, Gauge, Histogram
from functools import wraps
import time
# 定义应用级别的指标
REQUEST_COUNT = Counter('app_requests_total', '请求总数', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('app_request_duration_seconds', '请求耗时', ['method', 'endpoint'])
ACTIVE_USERS = Gauge('app_active_users', '活跃用户数')
DB_QUERY_COUNT = Counter('app_db_queries_total', '数据库查询总数')
def monitor_request(func):
"""监控请求的装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
# 这里可以获取请求信息
# 在实际框架中,可以从request对象获取
method = 'GET' # 示例
endpoint = func.__name__
try:
result = func(*args, **kwargs)
status = '200'
return result
except Exception as e:
status = '500'
raise e
finally:
# 记录指标
duration = time.time() - start_time
REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status).inc()
REQUEST_LATENCY.labels(method=method, endpoint=endpoint).observe(duration)
return wrapper
# 使用示例
@monitor_request
def process_order(order_data):
"""处理订单"""
# 业务逻辑
DB_QUERY_COUNT.inc() # 记录数据库查询
return {"status": "success"}
使用装饰器可以让监控代码更加整洁,避免业务逻辑和监控代码混在一起。
9. 总结
通过这篇教程,我们完成了一个完整的Python监控告警系统:
- 环境搭建:使用Miniconda-Python3.11镜像快速创建开发环境
- 指标暴露:用
prometheus-client库暴露系统指标和应用指标 - 数据采集:部署Prometheus,定期抓取指标数据
- 告警设置:定义告警规则,在问题发生时自动检测
- 通知配置:通过Alertmanager发送告警通知
这个系统虽然简单,但包含了监控告警的核心要素。你可以在此基础上继续扩展:
- 添加更多监控指标(数据库连接数、缓存命中率、消息队列长度等)
- 集成Grafana创建漂亮的仪表盘
- 监控多个服务的依赖关系
- 实现自动化的故障恢复
监控告警不是一劳永逸的工作,而是一个持续优化的过程。开始时可以简单一些,先监控最重要的指标,然后根据实际运行情况逐步调整。
最重要的是,你的监控系统要能真正发现问题,并且在问题影响用户之前通知到你。这样你就能睡个安稳觉,不用担心半夜被叫起来处理故障了。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐




所有评论(0)