半夜三点手机在企业微信上开始疯狂震动。我迷迷糊糊摸到手机看到告警群里一行红字“生产服务 5xx 错误率超过阈值”。我一个激灵坐起来打开电脑登录 Grafana 一看——错误率确实超了但持续了不到两分钟就自己恢复了业务根本没受影响。这种剧本写过 FastAPI 后端服务的朋友应该都不陌生。折腾告警这事儿从 FastAPI 项目上线那天起就没消停过。一开始不配告警出了问题全靠用户反馈后来配了告警又天天被无效告警骚扰。这中间的平衡点我摸索了大半年。这篇文章就聊聊告警到底怎么搞从服务侧埋点到企业微信机器人推送再到去重、静默、恢复通知一套能落地的方案完整讲清楚。适合正在用 FastAPI 做服务、又不想被告警工具本身折磨的后端开发者。1. 为什么你的告警总在半夜炸醒你1.1 先想清楚告警不是越多越好大部分人搭告警系统第一个念头就是“把能监控的都监控起来”。CPU 超了告警、内存超了告警、接口慢了告警、磁盘满了告警、进程挂了告警一天下来群里几百条消息。结果是什么告警变成了背景噪音真正出大事的时候反而没人看了。我见过最夸张的一次是某个服务磁盘使用率超过 85% 触发告警但因为磁盘一直缓慢增长这个告警每隔 5 分钟就重复推一次连续推了一整夜。值班同学早就把企业微信静音了结果第二天凌晨数据库真的因为磁盘写满宕机了反而没人第一时间发现。这就是告警设计的第一原则告警是给人看的不是给机器人看的。每条告警都必须回答三个问题——现在发生了什么、影响什么业务、值班的人该怎么处理。回答不了这三个问题的告警不如不配。所以我更愿意把告警分成三类事故类、预警类、通知类。事故类必须立刻处理比如接口大量 5xx、服务进程消失、数据库连接池打满预警类需要关注但不至于半夜捞人比如磁盘使用率超过 70%、P95 延迟连续 15 分钟升高通知类只是同步状态比如服务发版成功、配置变更完成。三类告警的推送渠道、响应的紧急程度、重复推送的策略完全不同。1.2 分级、分渠道、分对象告警规则怎么定在小团队里告警分级不用搞得很复杂建议直接按 P0、P1、P2 三档来设计。P0 是最高级别指直接影响用户可用性的事故比如核心接口错误率飙升、服务完全不可用、数据库连接失败。P0 告警必须通过电话或者企业微信机器人强提醒而且要有升级机制发出去 15 分钟没人确认就自动通知到技术负责人。P1 是严重但还没到全面瘫痪的程度比如某个非核心接口延迟变高、缓存命中率骤降这类可以推送到告警群要求值班人员在 30 分钟内响应。P2 是警告和通知比如磁盘使用率超过阈值、离线任务重试成功这类白天看看就行不需要半夜打搅人。这么分级带来的直接变化是值班的人终于知道哪些消息需要立刻从床上爬起来哪些可以等天亮再处理。分级明确之后告警的推送策略也必须跟着区分不能所有告警都用同一个 webhook 推到同一个群否则分级就失去了意义。建议至少分成两个群一个“紧急告警”群只有 P0 和 P1 能进一个“日常巡检”群P2 和通知类消息进去。这样处理告警的人不用在几百条消息里翻找哪个是真正要紧的。2. FastAPI服务侧埋点健康检查与指标暴露2.1 双探针设计/healthz 和 /readyz 要分开很多 FastAPI 项目的健康检查就一个/health接口返回一个{status: ok}就完事了。看起来好像没什么问题但实际上这个接口只能证明你uvicorn进程还活着什么也证明不了。我推荐把健康检查拆成两个探针存活探针/healthz和就绪探针/readyz。存活探针的作用很纯粹就是告诉负载均衡器和监控系统“进程还活着”。这个接口什么都不要查直接返回 200 就行因为只要进程活着它就该通过。就绪探针则要检查服务的外部依赖数据库能不能连上、Redis 能不能读写、消息队列有没有积压。只有就绪探针通过了流量才应该打到这个实例上。FastAPI 里实现这两个接口非常简单import time from fastapi import FastAPI from redis import asyncio as aioredis from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine app FastAPI(titledemo-service) # 存活探针只证明进程活着 app.get(/healthz) async def healthz(): return {status: ok, time: int(time.time())} # 就绪探针检查所有关键依赖 app.get(/readyz) async def readyz(): checks {} ok True try: # 检查数据库SQLAlchemy 异步引擎 async with app.state.db_engine.connect() as conn: await conn.execute(text(SELECT 1)) checks[database] ok except Exception as e: checks[database] ferror: {e} ok False try: r app.state.redis await r.ping() checks[redis] ok except Exception as e: checks[redis] ferror: {e} ok False return {ready: ok, checks: checks}, 200 if ok else 503注意就绪探针返回 503 而不是让请求超时。这里有个坑如果就绪探针本身执行得太慢比如数据库连接超时设了 30 秒那负载均衡器可能会判定这个实例已经死掉并把它摘除但实际上服务只是依赖出了问题。所以就绪探针里的每个依赖检查都要用短超时一般 2 到 3 秒就够了超过就认为失败。2.2 用Prometheus指标暴露业务水位健康检查只能告诉你“现在还活着”但一个服务从“正常”到“挂掉”之间往往有一段很长的恶化过程。磁盘在慢慢变满接口在慢慢变慢错误率在慢慢上升。这些趋势性的信号需要靠指标来捕捉。FastAPI 项目接入 Prometheus 指标不算复杂只需要在启动时创建若干指标对象再挂一个中间件统计请求数据from fastapi import FastAPI, Request from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST from starlette.responses import Response from starlette.middleware.base import BaseHTTPMiddleware import time REQUEST_COUNT Counter( http_requests_total, Total HTTP requests, [method, path, status] ) REQUEST_DURATION Histogram( http_request_duration_seconds, HTTP request duration, [method, path], buckets(0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10) ) class MetricsMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start time.perf_counter() try: response await call_next(request) except Exception: response Response(status_code500) duration time.perf_counter() - start path request.url.path # 过滤指标接口本身避免循环采集 if path not in (/metrics, /healthz, /readyz): REQUEST_COUNT.labels( methodrequest.method, pathpath, statusresponse.status_code ).inc() REQUEST_DURATION.labels( methodrequest.method, pathpath ).observe(duration) return response app.add_middleware(MetricsMiddleware) app.get(/metrics) async def metrics(): return Response(contentgenerate_latest(), media_typeCONTENT_TYPE_LATEST)这个中间件会把每个请求的方法、路径、状态码都记录下来。Grafana 那边导入 vCenter 告警也好、对接 Prometheus 数据源也好最终都是靠这类指标来出图和出告警规则。指标一旦通了后面写告警规则就非常灵活。这里有一个很关键的细节/metrics接口本身不能被隔离在认证之外但也不能暴露到公网。建议要么让 Prometheus 直接在内网抓取要么给/metrics加一层简单的 Token 认证防止别人把你的指标数据拉走。很多小项目图省事把/metrics暴露到公网结果被人刷接口反而成了隐患。2.3 业务告警到底要看哪些指标不少 FastAPI 项目把 Prometheus 指标埋了Grafana 面板也配得挺好看但真正写告警规则的时候不知道从哪下手。这里分享一下我总结的最低限度告警指标集。第一类是基础设施指标CPU 使用率、内存使用率、磁盘使用率、网络出入流量。这类指标用 node-exporter 之类的工具就能采集不一定需要业务埋点。但注意阈值别拍脑袋定磁盘 85% 告警、内存 90% 告警这些数字都要结合实际环境调整。第二类是应用指标请求 QPS、5xx 错误率、P95 延迟。这两项可以直接在 FastAPI 中间件里统计。第三类是依赖指标数据库连接池使用率、Redis 命中率、消息队列积压数量。这些虽然来自外部系统但如果你的服务依赖它们就必须纳入告警范围。用 Grafana 展示这些指标时我习惯把“当前值、过去一小时趋势、昨天同一时段”放在同一个面板里。这样有一条告警过来你第一眼就能判断是突发问题还是缓慢恶化。3. 告警推送从企业微信机器人到消息去重3.1 企业微信机器人Webhook接入套路告警最终要触达人才算数。小团队最方便的方案就是企业微信群机器人——创建群聊、添加机器人、拿到 Webhook 地址往这个地址 POST 一段 JSON消息就进群了。机器人创建没什么好说的在群设置里添加“群机器人”就能拿到 Webhook。重点说一下消息格式建议直接用 markdown 类型因为可以在告警内容里用颜色区分级别、用加粗突出重点。import httpx import time WEBHOOK_URL https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key LEVEL_COLOR { P0: warning, P1: warning, P2: comment, } async def send_alert(level: str, title: str, content: str): color LEVEL_COLOR.get(level, comment) markdown_content ( f**font color{color}{level} 告警/font**\n f 标题**{title}**\n f {content}\n f 时间{time.strftime(%Y-%m-%d %H:%M:%S)}\n ) payload { msgtype: markdown, markdown: {content: markdown_content}, } async with httpx.AsyncClient(timeout5) as client: resp await client.post(WEBHOOK_URL, jsonpayload) data resp.json() if data.get(errcode) ! 0: # 这里必须留个后手Webhook 发送失败要能打到本地日志 print(fsend alert failed: {data})发送告警的这段代码要封装成一个独立模块后面所有告警规则都调它。另外强烈建议把 Webhook 地址放在环境变量里不要硬编码在代码中。就算代码仓库是私有的也不排除某天仓库被公开或者分享给别人Webhook 泄露出去意味着任何人都能往你的告警群发垃圾消息。企业微信机器人有一个限制需要关注每分钟最多发送 20 条消息。如果告警风暴发生很有可能会触发频率限制。所以推送模块里最好做一层本地队列把发送失败的告警缓存下来退避重试而不是直接把异常抛出去。3.2 告警去重与静默防止告警风暴的核心手段同一台服务器磁盘满了如果不做任何处理Prometheus 每隔 15 秒拉取一次指标每发现一次超阈值就触发一次告警Grafana 和 Alertmanager 如果配置不当企业微信机器人每分钟能收到 20 条重复消息直接把你的群淹没。解决重复告警的标准方案是“状态机 静默期”。每个告警规则本质上是一个状态机正常ok和触发firing两个状态。只有状态从 ok 切换到 firing 的时候才发送告警在 firing 状态下即使每次检查都超阈值也只按静默周期发送而不是每次都发。下面是一个轻量级的状态机实现不依赖外部存储适合单机部署的小服务import time class AlertStateMachine: def __init__(self, rule_name, dedup_seconds300, on_sendNone): self.rule_name rule_name self.dedup_seconds dedup_seconds # 静默期单位秒 self.state ok self.last_send_at 0 self.firing_since None self._on_send on_send def evaluate(self, current_status: bool): now time.time() if current_status: # 触发条件成立 if self.state ok: # ok - firing发送告警 self.state firing self.firing_since now self._send() else: # 已经在 firing检查是否过了静默期 if now - self.last_send_at self.dedup_seconds: self._send() else: # 触发条件不成立 if self.state firing: # firing - ok发送恢复通知 self.state ok self.firing_since None self._on_send(recover, self.rule_name) def _send(self): self.last_send_at time.time() if self._on_send: self._on_send(alert, self.rule_name)这个状态机的逻辑看起来简单但非常重要的一点是恢复通知也必须走同一个状态机不能只在触发时发告警、恢复时不通知。否则值班的人收到一条告警后根本不知道这个问题是否已经解决只能自己去查。我见过不少团队就是因为没有恢复通知导致值班人员每天都要在 Grafana 上手动确认“这个问题现在好了没”效率非常低下。静默期的取值也要结合业务特点。像磁盘使用率这种缓慢变化的指标静默期可以设 3600 秒一小时重复一次已经足够而接口 5xx 这种突发性较强的指标静默期最好短一些比如 300 秒确保在故障持续期间能够持续提醒。3.3 升级机制告警发出去没人理怎么办告警真正让人头大的情况不是没人收到而是收到的人看了一眼、然后又睡过去了。如果你用的是企业微信机器人消息已读不可控所以必须做“升级机制”。简单的做法是在状态机里记录首次触发时间如果在设定时间内没有收到确认就升级到更高一级的报障渠道。比如 P0 告警发出后 15 分钟内没有人在群里回复“处理中”自动调用电话接口打给值班负责人再过 10 分钟没人接就打到技术负责人那里。我自己的经验是哪怕只是加一层手动确认告警的响应时间都能缩短一半——因为大家都知道不确认会继续升级。等团队规模起来了还可以接入值班轮转表自动决定当前告警该找谁。小团队阶段就别自己造复杂的轮转系统了一个 Python 脚本根据日期算值班人把名字和手机号写进配置里就够用。4. FastAPI开发与部署环境的坑顺带救一个算一个4.1 用uv创建虚拟环境省心不止一点很多 FastAPI 项目环境问题根源都在虚拟环境管理。以前我用 pip venv 也能跑但新机器上配环境总是要费一番功夫。后来换成了 uv体验好了不止一个档次。uv 是一个用 Rust 写的 Python 包管理器最大的优点是快而且命令非常直观。创建一个 FastAPI 项目虚拟环境只需要几条命令uv venv .venv source .venv/bin/activate uv pip install fastapi uvicorn[standard] httpx prometheus-client也可以在项目根目录用uv init初始化然后uv add fastapi它会自动生成使用文档和锁文件依赖版本都被固定住换机器部署的时候不再出现“在我电脑上明明好好的”这种问题。之前我在 PyCharm 里新建项目时如果直接用全局 Python 解释器经常出现包装到别的环境、项目里却导入不了的情况。用 uv 创建完.venv之后在 PyCharm 设置里把解释器指到.venv/bin/python问题就解决了。4.2 热更新不生效八成不是FastAPI的锅不少朋友碰到“FastAPI 启动不热更新”的问题第一反应就是 FastAPI 本身有 Bug。其实 FastAPI 本身并不负责热更新这事儿是 uvicorn 的--reload参数在管。启动命令应该是uvicorn main:app --reload --reload-dir .--reload开启文件监听--reload-dir限定监听目录。如果你把项目放在 Docker 里运行还需要把代码目录挂载进容器挂载没配置好容器里的文件变了uvicorn 自然感知不到。另外如果监听目录太大uvicorn 的文件系统事件太多也可能会漏掉部分变化。还有一个小坑--reload参数只在开发环境用生产环境必须去掉。有些同学图省事生产也用--reload结果代码文件被意外改动时服务直接自动重启了这种你根本想不到的故障排查起来极其痛苦。4.3 PyCharm安装FastAPI失败的排查思路PyCharm 里安装 FastAPI 报错最常见的原因是解释器选错了。Python 2 的解释器装不了 FastAPI因为 FastAPI 只支持 Python 3.7 及以上版本。另一个常见原因是 pip 默认源在某些网络环境下访问不稳定导致安装超时。可以换成国内镜像pip install fastapi -i https://pypi.tuna.tsinghua.edu.cn/simple如果提示权限错误多半是当前 Python 环境是系统级环境没有写权限。解决办法就是给项目创建一个虚拟环境不要在系统级环境里乱装包装。还有一个大家容易忽略的PyCharm 里新建 FastAPI 项目时如果选了“New environment using Virtualenv”但 Python 版本选的还是系统自带的 3.6那装 FastAPI 照样会报版本不兼容。在 PyCharm 的设置界面里确认一下解释器版本比看一长串报错日志更高效。5. 告警系统上线前先给自己做一次自检5.1 必须回答的十个问题告警系统搭建好之后先别急着扔进生产先过一遍下面这份自检清单。问题合格标准每条告警是否有明确负责人群里能查到当前值班人电话告警消息是否包含影响范围能说出“哪个服务、哪个接口、什么错误”是否有恢复通知故障解除后必须收到 ok 消息同一故障是否会重复报警静默期内不会连续轰炸告警分级是否正确P0 不会发给 P2 的群阈值是否经过至少一周验证没有出现每天误报或漏报是否有升级机制超时未处理会自动通知更高一级Webhook 是否通过环境变量注入源码里找不到明文 key告警模块是否有自身失败兜底Webhook 发送失败要能记录日志告警系统是否有维护文档新成员加入能看懂告警含义这份清单不是凭空列出来的每一条背后都对应我踩过的坑。比如“阈值是否经过验证”这条最开始我把 5xx 错误率阈值设为 1%结果某个服务因为爬虫扫描每天凌晨都会触发几次 502导致告警天天响。后来把阈值调成 5%、并且要求持续 5 分钟才触发世界才清净。5.2 我的一次误删告警规则经历自检还要覆盖一个场景告警规则本身的变更管理。有一次我为了调试某条规则直接在 Grafana 的 Alerting 页面上把规则临时禁用调试完成后忘了重新启用。结果半夜服务真的出故障了整整一个小时没人发现直到用户打电话过来反馈。那次事故给我的教训是告警规则的变更也必须走审批流程禁用了哪条规则、为什么禁用、什么时候恢复都要记下来。后来我干脆给团队定了一个规矩——所有告警规则的增删改都要提交到代码仓库通过配置中心统一管理。Prometheus 的告警规则文件本身就是文本完全可以用 Git 来做版本管理。每次改动都有 diff出了问题可以随时回滚。Grafana 的 Alerting 规则也能通过 provisioning 方式管理好处是环境迁移的时候能批量部署不用在网页上一个一个点。配置中心听起来很重但对小团队来说一个 Git 仓库加几张 YAML 文件就足够了。至少需要有人改了告警规则其他人 review 时能看出来而不是半夜看到一条新告警一脸懵地研究这规则是谁配的。5.3 给 FastAPI 项目加一个“告警模块自检”最后分享一个很实用的小功能给 FastAPI 项目增加一个POST /internal/alert-test接口只允许内网访问用来手动触发测试告警。每次变更告警配置后调一下这个接口确认企业微信机器人能收到、格式没问题。这个接口看起来有点多余但真到了出故障的时候你才能确定告警链路是通的。这个自检接口还能承担一个任务告警模块本身的错误上报。如果推送代码出现异常至少能把这个异常记录到本地日志文件或者发到另一个独立的“系统消息”群。告警系统自己出问题却没被发现才是最容易翻车的地方。结尾一点真实体会搞了这么久的告警我最深的感触是告警系统的最终目标不是把监控做得多全而是让你在收到告警的时候能快速判断“这事儿重不重要、该找谁、现在能做什么”。它应该是一个辅助决策的工具而不是一个半夜打电话催你起床的闹钟。如果你现在正被一堆无效告警折磨我的建议是从今天晚上开始做减法。把能关的告警规则先停一半留下一周观察看看真正有价值的有几条再把阈值和静默期慢慢调到位。告警设计这东西没有标准答案只有适不适合你的业务场景。折腾一段时间之后你会发现能安安静静睡到天亮才是告警系统最合格的成绩。
FastAPI服务告警实战：从健康检查到企业微信机器人推送
半夜三点手机在企业微信上开始疯狂震动。我迷迷糊糊摸到手机看到告警群里一行红字“生产服务 5xx 错误率超过阈值”。我一个激灵坐起来打开电脑登录 Grafana 一看——错误率确实超了但持续了不到两分钟就自己恢复了业务根本没受影响。这种剧本写过 FastAPI 后端服务的朋友应该都不陌生。折腾告警这事儿从 FastAPI 项目上线那天起就没消停过。一开始不配告警出了问题全靠用户反馈后来配了告警又天天被无效告警骚扰。这中间的平衡点我摸索了大半年。这篇文章就聊聊告警到底怎么搞从服务侧埋点到企业微信机器人推送再到去重、静默、恢复通知一套能落地的方案完整讲清楚。适合正在用 FastAPI 做服务、又不想被告警工具本身折磨的后端开发者。1. 为什么你的告警总在半夜炸醒你1.1 先想清楚告警不是越多越好大部分人搭告警系统第一个念头就是“把能监控的都监控起来”。CPU 超了告警、内存超了告警、接口慢了告警、磁盘满了告警、进程挂了告警一天下来群里几百条消息。结果是什么告警变成了背景噪音真正出大事的时候反而没人看了。我见过最夸张的一次是某个服务磁盘使用率超过 85% 触发告警但因为磁盘一直缓慢增长这个告警每隔 5 分钟就重复推一次连续推了一整夜。值班同学早就把企业微信静音了结果第二天凌晨数据库真的因为磁盘写满宕机了反而没人第一时间发现。这就是告警设计的第一原则告警是给人看的不是给机器人看的。每条告警都必须回答三个问题——现在发生了什么、影响什么业务、值班的人该怎么处理。回答不了这三个问题的告警不如不配。所以我更愿意把告警分成三类事故类、预警类、通知类。事故类必须立刻处理比如接口大量 5xx、服务进程消失、数据库连接池打满预警类需要关注但不至于半夜捞人比如磁盘使用率超过 70%、P95 延迟连续 15 分钟升高通知类只是同步状态比如服务发版成功、配置变更完成。三类告警的推送渠道、响应的紧急程度、重复推送的策略完全不同。1.2 分级、分渠道、分对象告警规则怎么定在小团队里告警分级不用搞得很复杂建议直接按 P0、P1、P2 三档来设计。P0 是最高级别指直接影响用户可用性的事故比如核心接口错误率飙升、服务完全不可用、数据库连接失败。P0 告警必须通过电话或者企业微信机器人强提醒而且要有升级机制发出去 15 分钟没人确认就自动通知到技术负责人。P1 是严重但还没到全面瘫痪的程度比如某个非核心接口延迟变高、缓存命中率骤降这类可以推送到告警群要求值班人员在 30 分钟内响应。P2 是警告和通知比如磁盘使用率超过阈值、离线任务重试成功这类白天看看就行不需要半夜打搅人。这么分级带来的直接变化是值班的人终于知道哪些消息需要立刻从床上爬起来哪些可以等天亮再处理。分级明确之后告警的推送策略也必须跟着区分不能所有告警都用同一个 webhook 推到同一个群否则分级就失去了意义。建议至少分成两个群一个“紧急告警”群只有 P0 和 P1 能进一个“日常巡检”群P2 和通知类消息进去。这样处理告警的人不用在几百条消息里翻找哪个是真正要紧的。2. FastAPI服务侧埋点健康检查与指标暴露2.1 双探针设计/healthz 和 /readyz 要分开很多 FastAPI 项目的健康检查就一个/health接口返回一个{status: ok}就完事了。看起来好像没什么问题但实际上这个接口只能证明你uvicorn进程还活着什么也证明不了。我推荐把健康检查拆成两个探针存活探针/healthz和就绪探针/readyz。存活探针的作用很纯粹就是告诉负载均衡器和监控系统“进程还活着”。这个接口什么都不要查直接返回 200 就行因为只要进程活着它就该通过。就绪探针则要检查服务的外部依赖数据库能不能连上、Redis 能不能读写、消息队列有没有积压。只有就绪探针通过了流量才应该打到这个实例上。FastAPI 里实现这两个接口非常简单import time from fastapi import FastAPI from redis import asyncio as aioredis from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine app FastAPI(titledemo-service) # 存活探针只证明进程活着 app.get(/healthz) async def healthz(): return {status: ok, time: int(time.time())} # 就绪探针检查所有关键依赖 app.get(/readyz) async def readyz(): checks {} ok True try: # 检查数据库SQLAlchemy 异步引擎 async with app.state.db_engine.connect() as conn: await conn.execute(text(SELECT 1)) checks[database] ok except Exception as e: checks[database] ferror: {e} ok False try: r app.state.redis await r.ping() checks[redis] ok except Exception as e: checks[redis] ferror: {e} ok False return {ready: ok, checks: checks}, 200 if ok else 503注意就绪探针返回 503 而不是让请求超时。这里有个坑如果就绪探针本身执行得太慢比如数据库连接超时设了 30 秒那负载均衡器可能会判定这个实例已经死掉并把它摘除但实际上服务只是依赖出了问题。所以就绪探针里的每个依赖检查都要用短超时一般 2 到 3 秒就够了超过就认为失败。2.2 用Prometheus指标暴露业务水位健康检查只能告诉你“现在还活着”但一个服务从“正常”到“挂掉”之间往往有一段很长的恶化过程。磁盘在慢慢变满接口在慢慢变慢错误率在慢慢上升。这些趋势性的信号需要靠指标来捕捉。FastAPI 项目接入 Prometheus 指标不算复杂只需要在启动时创建若干指标对象再挂一个中间件统计请求数据from fastapi import FastAPI, Request from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST from starlette.responses import Response from starlette.middleware.base import BaseHTTPMiddleware import time REQUEST_COUNT Counter( http_requests_total, Total HTTP requests, [method, path, status] ) REQUEST_DURATION Histogram( http_request_duration_seconds, HTTP request duration, [method, path], buckets(0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10) ) class MetricsMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start time.perf_counter() try: response await call_next(request) except Exception: response Response(status_code500) duration time.perf_counter() - start path request.url.path # 过滤指标接口本身避免循环采集 if path not in (/metrics, /healthz, /readyz): REQUEST_COUNT.labels( methodrequest.method, pathpath, statusresponse.status_code ).inc() REQUEST_DURATION.labels( methodrequest.method, pathpath ).observe(duration) return response app.add_middleware(MetricsMiddleware) app.get(/metrics) async def metrics(): return Response(contentgenerate_latest(), media_typeCONTENT_TYPE_LATEST)这个中间件会把每个请求的方法、路径、状态码都记录下来。Grafana 那边导入 vCenter 告警也好、对接 Prometheus 数据源也好最终都是靠这类指标来出图和出告警规则。指标一旦通了后面写告警规则就非常灵活。这里有一个很关键的细节/metrics接口本身不能被隔离在认证之外但也不能暴露到公网。建议要么让 Prometheus 直接在内网抓取要么给/metrics加一层简单的 Token 认证防止别人把你的指标数据拉走。很多小项目图省事把/metrics暴露到公网结果被人刷接口反而成了隐患。2.3 业务告警到底要看哪些指标不少 FastAPI 项目把 Prometheus 指标埋了Grafana 面板也配得挺好看但真正写告警规则的时候不知道从哪下手。这里分享一下我总结的最低限度告警指标集。第一类是基础设施指标CPU 使用率、内存使用率、磁盘使用率、网络出入流量。这类指标用 node-exporter 之类的工具就能采集不一定需要业务埋点。但注意阈值别拍脑袋定磁盘 85% 告警、内存 90% 告警这些数字都要结合实际环境调整。第二类是应用指标请求 QPS、5xx 错误率、P95 延迟。这两项可以直接在 FastAPI 中间件里统计。第三类是依赖指标数据库连接池使用率、Redis 命中率、消息队列积压数量。这些虽然来自外部系统但如果你的服务依赖它们就必须纳入告警范围。用 Grafana 展示这些指标时我习惯把“当前值、过去一小时趋势、昨天同一时段”放在同一个面板里。这样有一条告警过来你第一眼就能判断是突发问题还是缓慢恶化。3. 告警推送从企业微信机器人到消息去重3.1 企业微信机器人Webhook接入套路告警最终要触达人才算数。小团队最方便的方案就是企业微信群机器人——创建群聊、添加机器人、拿到 Webhook 地址往这个地址 POST 一段 JSON消息就进群了。机器人创建没什么好说的在群设置里添加“群机器人”就能拿到 Webhook。重点说一下消息格式建议直接用 markdown 类型因为可以在告警内容里用颜色区分级别、用加粗突出重点。import httpx import time WEBHOOK_URL https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key LEVEL_COLOR { P0: warning, P1: warning, P2: comment, } async def send_alert(level: str, title: str, content: str): color LEVEL_COLOR.get(level, comment) markdown_content ( f**font color{color}{level} 告警/font**\n f 标题**{title}**\n f {content}\n f 时间{time.strftime(%Y-%m-%d %H:%M:%S)}\n ) payload { msgtype: markdown, markdown: {content: markdown_content}, } async with httpx.AsyncClient(timeout5) as client: resp await client.post(WEBHOOK_URL, jsonpayload) data resp.json() if data.get(errcode) ! 0: # 这里必须留个后手Webhook 发送失败要能打到本地日志 print(fsend alert failed: {data})发送告警的这段代码要封装成一个独立模块后面所有告警规则都调它。另外强烈建议把 Webhook 地址放在环境变量里不要硬编码在代码中。就算代码仓库是私有的也不排除某天仓库被公开或者分享给别人Webhook 泄露出去意味着任何人都能往你的告警群发垃圾消息。企业微信机器人有一个限制需要关注每分钟最多发送 20 条消息。如果告警风暴发生很有可能会触发频率限制。所以推送模块里最好做一层本地队列把发送失败的告警缓存下来退避重试而不是直接把异常抛出去。3.2 告警去重与静默防止告警风暴的核心手段同一台服务器磁盘满了如果不做任何处理Prometheus 每隔 15 秒拉取一次指标每发现一次超阈值就触发一次告警Grafana 和 Alertmanager 如果配置不当企业微信机器人每分钟能收到 20 条重复消息直接把你的群淹没。解决重复告警的标准方案是“状态机 静默期”。每个告警规则本质上是一个状态机正常ok和触发firing两个状态。只有状态从 ok 切换到 firing 的时候才发送告警在 firing 状态下即使每次检查都超阈值也只按静默周期发送而不是每次都发。下面是一个轻量级的状态机实现不依赖外部存储适合单机部署的小服务import time class AlertStateMachine: def __init__(self, rule_name, dedup_seconds300, on_sendNone): self.rule_name rule_name self.dedup_seconds dedup_seconds # 静默期单位秒 self.state ok self.last_send_at 0 self.firing_since None self._on_send on_send def evaluate(self, current_status: bool): now time.time() if current_status: # 触发条件成立 if self.state ok: # ok - firing发送告警 self.state firing self.firing_since now self._send() else: # 已经在 firing检查是否过了静默期 if now - self.last_send_at self.dedup_seconds: self._send() else: # 触发条件不成立 if self.state firing: # firing - ok发送恢复通知 self.state ok self.firing_since None self._on_send(recover, self.rule_name) def _send(self): self.last_send_at time.time() if self._on_send: self._on_send(alert, self.rule_name)这个状态机的逻辑看起来简单但非常重要的一点是恢复通知也必须走同一个状态机不能只在触发时发告警、恢复时不通知。否则值班的人收到一条告警后根本不知道这个问题是否已经解决只能自己去查。我见过不少团队就是因为没有恢复通知导致值班人员每天都要在 Grafana 上手动确认“这个问题现在好了没”效率非常低下。静默期的取值也要结合业务特点。像磁盘使用率这种缓慢变化的指标静默期可以设 3600 秒一小时重复一次已经足够而接口 5xx 这种突发性较强的指标静默期最好短一些比如 300 秒确保在故障持续期间能够持续提醒。3.3 升级机制告警发出去没人理怎么办告警真正让人头大的情况不是没人收到而是收到的人看了一眼、然后又睡过去了。如果你用的是企业微信机器人消息已读不可控所以必须做“升级机制”。简单的做法是在状态机里记录首次触发时间如果在设定时间内没有收到确认就升级到更高一级的报障渠道。比如 P0 告警发出后 15 分钟内没有人在群里回复“处理中”自动调用电话接口打给值班负责人再过 10 分钟没人接就打到技术负责人那里。我自己的经验是哪怕只是加一层手动确认告警的响应时间都能缩短一半——因为大家都知道不确认会继续升级。等团队规模起来了还可以接入值班轮转表自动决定当前告警该找谁。小团队阶段就别自己造复杂的轮转系统了一个 Python 脚本根据日期算值班人把名字和手机号写进配置里就够用。4. FastAPI开发与部署环境的坑顺带救一个算一个4.1 用uv创建虚拟环境省心不止一点很多 FastAPI 项目环境问题根源都在虚拟环境管理。以前我用 pip venv 也能跑但新机器上配环境总是要费一番功夫。后来换成了 uv体验好了不止一个档次。uv 是一个用 Rust 写的 Python 包管理器最大的优点是快而且命令非常直观。创建一个 FastAPI 项目虚拟环境只需要几条命令uv venv .venv source .venv/bin/activate uv pip install fastapi uvicorn[standard] httpx prometheus-client也可以在项目根目录用uv init初始化然后uv add fastapi它会自动生成使用文档和锁文件依赖版本都被固定住换机器部署的时候不再出现“在我电脑上明明好好的”这种问题。之前我在 PyCharm 里新建项目时如果直接用全局 Python 解释器经常出现包装到别的环境、项目里却导入不了的情况。用 uv 创建完.venv之后在 PyCharm 设置里把解释器指到.venv/bin/python问题就解决了。4.2 热更新不生效八成不是FastAPI的锅不少朋友碰到“FastAPI 启动不热更新”的问题第一反应就是 FastAPI 本身有 Bug。其实 FastAPI 本身并不负责热更新这事儿是 uvicorn 的--reload参数在管。启动命令应该是uvicorn main:app --reload --reload-dir .--reload开启文件监听--reload-dir限定监听目录。如果你把项目放在 Docker 里运行还需要把代码目录挂载进容器挂载没配置好容器里的文件变了uvicorn 自然感知不到。另外如果监听目录太大uvicorn 的文件系统事件太多也可能会漏掉部分变化。还有一个小坑--reload参数只在开发环境用生产环境必须去掉。有些同学图省事生产也用--reload结果代码文件被意外改动时服务直接自动重启了这种你根本想不到的故障排查起来极其痛苦。4.3 PyCharm安装FastAPI失败的排查思路PyCharm 里安装 FastAPI 报错最常见的原因是解释器选错了。Python 2 的解释器装不了 FastAPI因为 FastAPI 只支持 Python 3.7 及以上版本。另一个常见原因是 pip 默认源在某些网络环境下访问不稳定导致安装超时。可以换成国内镜像pip install fastapi -i https://pypi.tuna.tsinghua.edu.cn/simple如果提示权限错误多半是当前 Python 环境是系统级环境没有写权限。解决办法就是给项目创建一个虚拟环境不要在系统级环境里乱装包装。还有一个大家容易忽略的PyCharm 里新建 FastAPI 项目时如果选了“New environment using Virtualenv”但 Python 版本选的还是系统自带的 3.6那装 FastAPI 照样会报版本不兼容。在 PyCharm 的设置界面里确认一下解释器版本比看一长串报错日志更高效。5. 告警系统上线前先给自己做一次自检5.1 必须回答的十个问题告警系统搭建好之后先别急着扔进生产先过一遍下面这份自检清单。问题合格标准每条告警是否有明确负责人群里能查到当前值班人电话告警消息是否包含影响范围能说出“哪个服务、哪个接口、什么错误”是否有恢复通知故障解除后必须收到 ok 消息同一故障是否会重复报警静默期内不会连续轰炸告警分级是否正确P0 不会发给 P2 的群阈值是否经过至少一周验证没有出现每天误报或漏报是否有升级机制超时未处理会自动通知更高一级Webhook 是否通过环境变量注入源码里找不到明文 key告警模块是否有自身失败兜底Webhook 发送失败要能记录日志告警系统是否有维护文档新成员加入能看懂告警含义这份清单不是凭空列出来的每一条背后都对应我踩过的坑。比如“阈值是否经过验证”这条最开始我把 5xx 错误率阈值设为 1%结果某个服务因为爬虫扫描每天凌晨都会触发几次 502导致告警天天响。后来把阈值调成 5%、并且要求持续 5 分钟才触发世界才清净。5.2 我的一次误删告警规则经历自检还要覆盖一个场景告警规则本身的变更管理。有一次我为了调试某条规则直接在 Grafana 的 Alerting 页面上把规则临时禁用调试完成后忘了重新启用。结果半夜服务真的出故障了整整一个小时没人发现直到用户打电话过来反馈。那次事故给我的教训是告警规则的变更也必须走审批流程禁用了哪条规则、为什么禁用、什么时候恢复都要记下来。后来我干脆给团队定了一个规矩——所有告警规则的增删改都要提交到代码仓库通过配置中心统一管理。Prometheus 的告警规则文件本身就是文本完全可以用 Git 来做版本管理。每次改动都有 diff出了问题可以随时回滚。Grafana 的 Alerting 规则也能通过 provisioning 方式管理好处是环境迁移的时候能批量部署不用在网页上一个一个点。配置中心听起来很重但对小团队来说一个 Git 仓库加几张 YAML 文件就足够了。至少需要有人改了告警规则其他人 review 时能看出来而不是半夜看到一条新告警一脸懵地研究这规则是谁配的。5.3 给 FastAPI 项目加一个“告警模块自检”最后分享一个很实用的小功能给 FastAPI 项目增加一个POST /internal/alert-test接口只允许内网访问用来手动触发测试告警。每次变更告警配置后调一下这个接口确认企业微信机器人能收到、格式没问题。这个接口看起来有点多余但真到了出故障的时候你才能确定告警链路是通的。这个自检接口还能承担一个任务告警模块本身的错误上报。如果推送代码出现异常至少能把这个异常记录到本地日志文件或者发到另一个独立的“系统消息”群。告警系统自己出问题却没被发现才是最容易翻车的地方。结尾一点真实体会搞了这么久的告警我最深的感触是告警系统的最终目标不是把监控做得多全而是让你在收到告警的时候能快速判断“这事儿重不重要、该找谁、现在能做什么”。它应该是一个辅助决策的工具而不是一个半夜打电话催你起床的闹钟。如果你现在正被一堆无效告警折磨我的建议是从今天晚上开始做减法。把能关的告警规则先停一半留下一周观察看看真正有价值的有几条再把阈值和静默期慢慢调到位。告警设计这东西没有标准答案只有适不适合你的业务场景。折腾一段时间之后你会发现能安安静静睡到天亮才是告警系统最合格的成绩。
