先说一个我踩过的坑：某次排查一个"偶尔卡 3 秒"的接口，翻遍了自己在after_request里打的耗时日志，最慢的一条只有 180ms。后来才反应过来——那些真正卡住的请求，日志压根就没打出来。原因是视图里抛了未捕获的异常，after_request根本没执行。从那次之后，我对"flask 请求时间记录"这件事的态度彻底变了：计时不是重点，保证每一条请求都留下一条可对账的记录才是重点。围绕这个核心，日志处理又牵出一堆问题——字段怎么设计、请求怎么串起来、多进程切割怎么不丢、慢请求怎么判定。这篇就把我在几个真实项目里趟出来的这套方案完整拆开讲，从计时点的选择一直到线上多 worker 的落盘策略，代码都可以直接抄。
1. after_request 方案为什么会在线上丢数据
网上讲 Flask 请求耗时记录的帖子，十篇有九篇是同一个写法：before_request里存个time.time()，after_request里减一下，print出来。这个写法在本地跑 demo 没问题，放到有真实流量、有异常、有流式响应、多 worker 的环境里，会同时暴露出三种失真。三种失真的成因完全不同，得分开处理。
1.1 计时点选在哪儿：perf_counter 和 time 的区别不只是精度
先说最基础的一个选择。用time.time()计时，本质上是读系统墙上时钟，而墙上时钟是会被调整的。NTP 同步、容器迁移、宿主机时钟漂移，任何一种都可能导致两个时间点之差是负数，或者凭空多出几百毫秒。我在一个跑在虚拟化环境里的服务上见过耗时字段出现-0.002s的记录，最后查出来是宿主机做了时钟校正。
time.perf_counter()读的是单调递增的高分辨率计数器，不受系统时间调整影响，唯一用途就是算差值。它不能用来表示"事件发生的绝对时间"，也不能跨机器比较——所以日志里那个可读的时间戳仍然要用datetime.now()，而耗时计算用perf_counter，两者分工明确：
import time from datetime import datetime from flask import g, request @app.before_request def _start_timer(): g._t0 = time.perf_counter() g._started_at = datetime.now()第二个问题更隐蔽：计时点到底覆盖了什么。before_request执行的时候，WSGI 服务器已经把 environ 组装好了，路由匹配也完成了，请求体还没读。所以从before_request起算的耗时，不含 TLS 握手、不含反向代理排队、不含请求体接收、不含响应传输。它衡量的其实是"框架内部处理时长"，这个指标本身有价值，但你得清楚它的边界，别拿它去回答"用户等了多久"这个问题。
想要更完整的视角，就把计时器提到 WSGI 层。用中间件包一层能拿到从进入应用到响应体全部吐完的时间：
class TimingMiddleware: def __init__(self, wsgi_app, callback): self.wsgi_app = wsgi_app self.callback = callback def __call__(self, environ, start_response): t0 = time.perf_counter() status_holder = {} def _start_response(status, headers, exc_info=None): status_holder["status"] = status return start_response(status, headers, exc_info) try: iterable = self.wsgi_app(environ, _start_response) except Exception: self.callback(environ, None, time.perf_counter() - t0) raise def _wrapper(): try: for chunk in iterable: yield chunk finally: if hasattr(iterable, "close"): iterable.close() self.callback( environ, status_holder.get("status"), time.perf_counter() - t0, ) return _wrapper() app.wsgi_app = TimingMiddleware(app.wsgi_app, on_request_done)这段代码的关键在_wrapper这个生成器。WSGI 应用返回的是一个可迭代对象，服务器拿到它之后才开始逐个 chunk 往外写。如果只在__call__里计时，流式响应（SSE、大文件下载、分页导出）测出来的只是"生成器创建耗时"，可能 1 毫秒都不到，而实际用户等了三十秒。把计时收尾放在生成器的finally里，才能覆盖到完整传输周期。finally里调close()也是必须的，否则生成器里的清理逻辑不会执行。
注意：中间件方案和
before_request方案不是二选一。我的做法是两个都留，中间件那个叫duration_total_ms，before_request那个叫duration_app_ms。当两者差距巨大时，基本可以直接判定是流式响应或者慢客户端在读，排查方向立刻就清楚了。
1.2 未捕获异常路径：after_request 确实会跳过
Flask 的请求分发流程大致是wsgi_app→full_dispatch_request→finalize_request→process_response（也就是触发after_request函数）。关键在于full_dispatch_request里视图抛异常之后，会被handle_user_exception接住：如果异常是 HTTP 异常就转成响应继续走，如果是普通异常就直接向上抛，交给wsgi_app里的handle_exception生成 500 响应。这条路径上finalize_request没有被调用，所以after_request一整个被跳过了。
真正一定会执行的是teardown_request，它在请求上下文弹出时触发，哪怕异常已经冒到最外层。所以兜底逻辑必须放在这里：
@app.teardown_request def _finalize_log(exc): if g.get("_logged"): return g._logged = True duration_ms = (time.perf_counter() - g.get("_t0", time.perf_counter())) * 1000 app.logger.error( "request failed", extra={ "method": request.method, "path": request.path, "duration_ms": round(duration_ms, 2), "error": repr(exc) if exc else None, }, exc_info=exc is not None, )_logged这个标记看着有点丑，但它解决的是幂等问题。正常请求走after_request时已经打过日志，teardown_request再来一次就重复了；异常请求走不到after_request，teardown_request补位。用小标记做互斥，比在两个地方各写一套判断条件要清楚得多。
这里还有个容易被忽略的点：teardown_request里绝对不能抛新异常。一旦这里出问题，Flask 会把它记到自己的错误日志里，然后原异常就被吞了，你看到的错误信息会完全对不上号。所以在 teardown 里访问request.path、g这些对象时，保险起见用getattr加默认值，别假设它们一定可用——比如请求在before_request之前就失败的极端情况。
1.3 静态文件、304 和流式响应：这些数字该怎么解读
日志量上来之后，你会发现占比最大的往往是静态资源请求。send_from_directory、favicon、健康检查探针，这些请求会以每秒几十上百条的频率把日志刷满，真正有价值的业务请求被埋在里面。我的处理方式是分层：健康检查（/healthz、/readyz）直接不记录或者降到 DEBUG 级别，静态资源只记一条极简格式，业务接口才记完整字段。
判断条件不要写死在路径字符串上，用request.endpoint：
SKIP_ENDPOINTS = {"static", "health.healthz", "health.readyz"} def _should_log(): return request.endpoint not in SKIP_ENDPOINTS流式响应的语义边界前面提过，这里补充一个实际会遇到的场景：send_file加了conditional=True之后，带If-None-Match的请求会直接返回 304，响应体是空的，耗时会非常小。如果你的慢请求统计里把 304 也算进去，分位数会被拉低，看起来一切正常，实际业务接口的慢请求被稀释了。所以聚合的时候一定要按 status 和 endpoint 分桶，别只看全局。
还有一种特别容易误判的情况：客户端主动断开连接。用户下载到一半关掉页面，WSGI 服务器那边会抛BrokenPipeError或者ClientDisconnected，这个异常在你的应用代码里完全看不到，但生成的是一次失败响应。如果你的报警依赖 5xx 数量，这类请求可能把报警打得震天响，实际服务器啥问题没有。
2. 一行日志里到底该有哪些字段
耗时算出来了，接下来是把它变成能被检索、能被聚合的东西。这里最忌讳的是随手app.logger.info(f"cost {x}ms")——纯字符串、没有结构，等到要做统计分析时，只能靠正则去啃，改一次格式就全崩。字段设计这一步值得提前花半小时想清楚，因为日志格式一旦上线，改动成本非常高。
2.1 字段清单和取舍逻辑
我把字段分成四组，每组的取舍理由不一样：
| 分组 | 字段 | 取舍理由 |
|---|---|---|
| 时间与定位 | ts、level、logger、request_id | request_id是串联同一次请求所有日志的唯一钥匙，必须有 |
| 请求特征 | method、path、endpoint、query_len | path用原始路径而非request.url，避免 query 里的敏感参数进日志 |
| 结果特征 | status、duration_ms、resp_bytes、exc_type | resp_bytes能帮你发现"慢但小"和"快但巨大"两类异常 |
| 客户端 | ip_hash、ua_short、uid | IP 做哈希或掩码，UA 截断到 100 字符，避免日志被撑爆 |
几个具体的坑值得单独说。
path和url的区别。request.url带完整 query string，而 query 里经常有 token、手机号、身份证号这类东西。有一个项目我就是图省事记了request.url，后来做日志导出给第三方分析时才发现问题，只能把整批日志删掉重来。用request.path记录路径，query 只记参数名不记值，这个习惯要从一开始就养成。
resp_bytes的获取。在after_request里通过response.calculate_content_length()拿，注意这个方法对流式响应返回None，因为它没法预知总长度。所以这个字段允许为空，不是 bug。
uid的取法要小心。如果你的应用有登录态，g.user_id在before_request阶段通常还没设置——鉴权逻辑一般写在视图的装饰器里，或者before_request的后半段。用一个before_request注册顺序的技巧解决：
@app.before_request def _early_timer(): g._t0 = time.perf_counter() # 鉴权装饰器 / 中间件在之后执行，设置 g.user_id # 打日志时用 g.get("user_id", "-") 兜底Flask 会按注册顺序执行before_request函数，所以把计时器注册在最前面（模块顶部）就能保证它第一个跑。但打日志的时候g.user_id是否已设置，取决于鉴权是否在请求处理路径上——很多应用是懒鉴权的（只有访问需要登录的接口才查），那大部分请求就是匿名，日志里出现-很正常，别当成异常数据。
2.2 用 dictConfig 接管日志，别跟 app.logger 纠缠
Flask 的app.logger行为在不同版本之间有过变化，默认情况下如果没手动配 handler，在非 debug 模式下app.logger.info(...)是没有输出的——这是个新手最容易困惑的点，明明写了日志，文件里就是空的。原因在于 Flask 默认只在 debug 模式给 logger 挂 handler，生产模式依赖于你通过标准 logging 配置来接。
官方现在推荐的路径是logging.config.dictConfig，用声明式配置一次性把 handler、formatter、filter、level 全定好：
import logging.config LOG_CONFIG = { "version": 1, "disable_existing_loggers": False, "filters": { "req_ctx": {"()": "myapp.logging_ext.RequestContextFilter"}, "drop_static": {"()": "myapp.logging_ext.SkipEndpointFilter"}, }, "formatters": { "plain": { "format": "%(asctime)s.%(msecs)03d %(levelname)-5s " "[%(request_id)s] %(name)s %(message)s", "datefmt": "%Y-%m-%dT%H:%M:%S", }, }, "handlers": { "stdout": { "class": "logging.StreamHandler", "formatter": "plain", "filters": ["req_ctx", "drop_static"], "stream": "ext://sys.stdout", }, }, "loggers": { "myapp": {"handlers": ["stdout"], "level": "INFO", "propagate": False}, "werkzeug": {"handlers": ["stdout"], "level": "WARNING", "propagate": False}, "urllib3": {"handlers": ["stdout"], "level": "WARNING", "propagate": False}, }, "root": {"handlers": ["stdout"], "level": "WARNING"}, }几个细节值得强调。
disable_existing_loggers: False必须显式写出来。默认值在 Python 3.8 之后是True，意思是会把配置生效之前已经创建的所有 logger 全部禁用。如果你的应用里某个第三方库在 import 阶段就建了 logger，配置一加载它就被静音了，之后这个库的所有日志你都看不到，排查问题时会有种"这库怎么一声不吭"的诡异感。
propagate: False要加在自己业务的 logger 上。不加的话，一条日志会沿着 logger 继承链往上，最后被 root 的 handler 再输出一次，日志文件里全是重复行。我见过有人用"去重"脚本处理这种情况，其实根因就是忘了关传播。
%(msecs)03d这个格式符在自定义datefmt时不会自动补零，得手动写03d，否则毫秒数可能是5而不是005，排序时会乱。
给werkzeug单独设WARNING很重要，否则开发模式下每个请求都会打一行访问日志，跟你的结构化日志重复了。生产环境用 gunicorn 时，访问日志走 gunicorn 的access_log_format，跟应用日志是两套，这个在后面单独讲。
2.3 JSON 和可读文本之间的折中方案
结构化日志最理想的形态是每行一个 JSON 对象，丢给日志聚合系统直接就能按字段检索。但纯 JSON 在本地开发时几乎没法看，一行几百个字符挤在一起，眼睛都花了。硬要选一边都不舒服，我实际采用的是按环境切换 formatter：本地用可读的 plain 格式，线上用 JSON。
import json, logging class JsonFormatter(logging.Formatter): RESERVED = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) def format(self, record): payload = { "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"), "level": record.levelname, "logger": record.name, "msg": record.getMessage(), } for k, v in record.__dict__.items(): if k not in self.RESERVED and not k.startswith("_"): payload[k] = v if record.exc_info: payload["exc"] = self.formatException(record.exc_info) return json.dumps(payload, ensure_ascii=False, default=str)这里有个细节：default=str不能省。日志里塞进去的任何非标准类型（Decimal、datetime、UUID、枚举）在json.dumps时都会抛TypeError，而这个异常发生在日志输出阶段，会把原本正常的业务流程打断。我见过因为这个参数漏写导致整个接口 500 的事故——日志写不出去，比日志写得不漂亮严重一百倍。
另一个坑是字段名冲突。用extra={}传自定义字段时，如果 key 跟LogRecord的内置属性重名（比如message、asctime、module、args），logging会直接抛KeyError。所以自定义字段统一加前缀，比如req_path、req_status、dur_ms，一眼就能跟内置属性区分开。
RESERVED那个集合的构造函数里我传了一堆空值，看着有点怪，但它能可靠地拿到当前 Python 版本下LogRecord的全部内置属性名，比自己硬编码一个列表要稳，跨版本升级也不容易漏。
3. request_id：把一次请求的所有日志串成一条线
只有一个耗时数字，排查问题时价值有限。真正有用的是：这个慢请求在业务代码里到底走了哪条分支、调了几次数据库、有没有重试。这些信息散落在不同模块的日志里，如果每行日志都带上同一个request_id，在日志系统里搜一次就能把整条链路拉出来。这是我认为投入产出比最高的一个改造。
3.1 生成策略与上游透传的信任边界
标准的做法是优先读请求头里的X-Request-Id，没有就自己生成一个。读上游的目的很实际：你的服务前面如果有一层网关或者另一个服务，让它们生成 ID 传下来，整条调用链就能串起来，而不是每跳都换一个新的。
import re, uuid from flask import g, request, Response RID_PATTERN = re.compile(r"^[A-Za-z0-9\-_]{8,64}$") @app.before_request def _assign_request_id(): incoming = request.headers.get("X-Request-Id", "") if RID_PATTERN.match(incoming): g.request_id = incoming else: g.request_id = uuid.uuid4().hex那个正则不是摆设。请求头是外部完全可控的输入，如果直接把任意长度、任意字符的内容写进日志，最直接的攻击方式是日志注入——传入带换行符的值，比如abc\n2026-01-01T00:00:00 INFO fake log entry，你的日志系统里就会凭空多出一条伪造的记录。在按行解析的日志管道里，这一条假记录可能引发一串误告警。限制字符集和长度（8 到 64 位）是最简单有效的防线。
顺带把 ID 回写到响应头，调用方排查问题的时候能直接对上：
@app.after_request def _echo_request_id(response): response.headers["X-Request-Id"] = g.get("request_id", "-") return response3.2 用 Filter 做无侵入注入
最省事的注入方式是logging.Filter。Filter 的职责本来是过滤，但它同时能在返回True之前修改record，等于是一个每行日志都会过的钩子：
from contextvars import ContextVar import logging _request_id_var = ContextVar("request_id", default="-") class RequestContextFilter(logging.Filter): def filter(self, record): record.request_id = _request_id_var.get() return True配合前面的before_request写入：
@app.before_request def _bind_context(): token = _request_id_var.set(g.request_id) g._rid_token = token用ContextVar而不是threading.local的理由很实际：ContextVar在 asyncio 和线程池场景下有正确的传播语义，而threading.local只在同一个线程内有效。Flask 的异步视图底层会切线程执行，用threading.local的话，你会在异步视图里看到request_id全是-。
绑定之后要记得解绑。Worker 进程是复用的，如果只 set 不 reset，同一个线程处理的下一个请求会继承上一个的值——当然下一个请求的before_request会重新 set，看起来没问题，但如果某个请求还没进before_request就打日志（比如 WSGI 层中间件的日志），拿到的就是上一个请求的 ID，串错链路比没有 ID 更糟糕。
@app.teardown_request def _unbind_context(exc): token = g.get("_rid_token") if token is not None: _request_id_var.reset(token)3.3 线程池、异步任务和后台作业里的上下文丢失
同步视图里这套机制工作得很好，但只要代码跨出请求线程，request_id就断了。常见的三种断开场景：
第一种，视图里ThreadPoolExecutor.submit(...)提交任务。submit不会自动复制 context，子线程里_request_id_var.get()返回默认值。正确写法是显式复制：
import contextvars ctx = contextvars.copy_context() executor.submit(ctx.run, my_task, arg1)第二种，run_in_executor。loop.run_in_executor同样不复制 context，得用asyncio.to_thread（Python 3.9+），后者内部会做copy_context()。
第三种，Celery 之类的任务队列。这个是完全脱离了请求生命周期的，唯一的办法是把request_id作为任务参数显式传下去，然后在任务开头重新set一次。别指望有什么魔法能自动带过去。
后台任务里带上了request_id之后，你的日志链路才真正完整：从前端一次点击，到网关，到应用，到异步任务，全部共享一个 ID。排查"用户说下单成功但没收到通知"这种问题时，这个能力的价值是压倒性的。
4. 多 worker 写同一个文件：切割那一刻开始出事
这是我最想重点讲的一节，因为它属于"配置写对了、本地测通了、上线跑几天才炸"的类型，而且炸得很隐蔽——日志不会报错，只是某些时间段的数据凭空消失了。
4.1 RotatingFileHandler 在多进程下的失效原理
logging.handlers.RotatingFileHandler的切割逻辑是：每次emit前检查当前文件大小，超过maxBytes就rename成.1、.2这样，然后重新打开一个新文件。
问题出在"每个进程各自检查"这件事上。假设有 4 个 gunicorn worker，各自持有一个指向同一个文件的文件描述符。当 A 进程判断需要切割，它把app.log重命名成app.log.1，然后创建新的app.log。此时 B、C、D 三个进程手里的文件描述符仍然指向那个已经被改名的 inode（也就是现在的app.log.1）。它们会继续往app.log.1里写，而且它们内部的文件大小计数器还停留在切割前的值，可能立刻又触发一次切割，把app.log.1覆盖掉。
最终表现就是：日志文件大小永远到不了你设的上限，内容却东一块西一块，有时一段时间的日志完全找不到。更麻烦的是，RotatingFileHandler在多进程下没有任何锁，两个进程同时切割会产生竞态，极端情况下直接抛FileNotFoundError，然后这个进程的日志 handler 就废了，后续所有日志静默丢失。
4.2 三种落盘策略的取舍
| 策略 | 适用场景 | 代价 |
|---|---|---|
| 只写 stdout，由运行环境收集 | 容器化部署（首选） | 依赖外部日志采集，本地调试要看终端 |
WatchedFileHandler+ 外部切割工具 | 传统虚拟机 / 物理机部署 | 需要配切割规则和信号通知 |
ConcurrentRotatingFileHandler（第三方库） | 单机多进程、又必须写本地文件 | 引入额外依赖，仍有锁竞争开销 |
容器化场景优先选第一种。应用只往 stdout 打，文件切割、轮转、归档全部交给容器运行时的日志驱动或者宿主机上的采集 agent。这样应用本身是无状态的，worker 数量随便调，水平扩容时也不会有多进程竞争的问题。这个方案唯一的代价是本地开发时得盯着终端看，加一个 "local" 环境配置切到文件 handler 就解决了。
传统部署必须写文件的话，用WatchedFileHandler。它的行为很朴素：每次emit前stat一下文件，如果发现 inode 变了（说明被外部工具重命名了），就关闭旧句柄重新打开。它自己不做切割，切割交给 logrotate 这类工具。logrotate的配置里create模式配合postrotate给进程发信号是最干净的，但很多 WSGI 服务器收到信号后只会重开自己的访问日志，不会重开应用层的 handler，所以最稳妥的还是把应用的WatchedFileHandler也指向同一个日志路径，让它自己检测 inode 变化。
第三种方案能用但不推荐。第三方库的做法是给文件加fcntl锁再操作，等于每次写日志都要抢一次锁，QPS 高的服务上这个开销不可忽视，而且锁粒度没处理好时会出现写阻塞，直接拖慢请求。
4.3 QueueHandler 削峰，以及它带来的新麻烦
不管写到哪里，日志 IO 都是同步的。磁盘慢的时候（尤其是某些网络挂载的卷），一次write加flush可能耗时几毫秒，高并发下这些时间会累加在请求响应时间里。标准库提供了QueueHandler+QueueListener的组合来解决：业务线程只往内存队列里put，一个独立的后台线程负责真正的写盘。
import logging, logging.handlers, queue log_queue = queue.Queue(maxsize=10000) qh = logging.handlers.QueueHandler(log_queue) root = logging.getLogger() root.addHandler(qh) file_handler = logging.handlers.WatchedFileHandler("/var/log/app/app.log") file_handler.setFormatter(json_formatter) listener = logging.handlers.QueueListener(log_queue, file_handler, respect_handler_level=True) listener.start()三个必须注意的点。
maxsize一定要设。不设的话队列无上限，磁盘卡住时内存会被日志条目吃光，最后触发 OOM。设了上限之后，队列满时的行为是阻塞put——也就是说日志压力会反向传导到业务线程。这个听起来很糟，但比 OOM 好。真正合理的做法是配合logging.raiseExceptions = False和一个自定义 handler，在队列满时丢弃并计数，用一个独立的指标告诉你有多少日志被丢了。
respect_handler_level=True要加上。默认情况下QueueListener会把队列里的所有记录无条件交给下游 handler，忽略了 handler 自己设的 level。加了这个参数，级别过滤才会按预期生效。
listener.start()之后要有对应的stop()。在进程退出时的钩子里调用，否则队列里剩余的日志会丢。atexit里注册一个就行。
注意：
QueueHandler解决的是单进程内的 IO 阻塞，完全不解决多进程写同一文件的问题。每个 worker 都有自己的队列和自己的QueueListener，也就都有自己的文件句柄，切割竞态一点没少。这两个问题是正交的，别混为一谈。
5. 慢请求的阈值该怎么定
有了耗时数据，接下来是判定。这一步最容易犯的错是凭感觉设一个数字——"超过 500ms 就是慢"。问题是，一个 PDF 导出接口的正常耗时可能是 3 秒，一个健康检查接口 50ms 都算慢。用同一个阈值套所有接口，结果只能是误报淹没有效信号，最后没人看报警了。
5.1 用分位数给每个接口单独定标
我的做法是：先全量记录两周，按endpoint分组算 P50 / P95 / P99，然后每个接口的告警阈值取max(P95 * 1.5, 业务可接受上限)。业务可接受上限这东西得跟产品确认，比如"订单列表页超过 800ms 用户体感就明显"，这就是硬约束，不能纯看数据。
具体落地时，日志里只记duration_ms和endpoint，聚合交给外部。如果不想上完整的日志分析栈，用prometheus_client的 Histogram 是最轻量的选择：
from prometheus_client import Histogram REQ_DURATION = Histogram( "http_request_duration_ms", "Request duration in milliseconds", ["endpoint", "method", "status"], buckets=(10, 25, 50, 100, 200, 400, 800, 1600, 3200, 6400), )分桶的边界要贴合实际的量级分布。如果所有请求都在 50ms 以内，桶全设成 1000 以上就没意义了，P50 永远落进第一个桶，算不出准确值。反过来，桶设得太密（比如每 10ms 一个），标签基数会爆炸，内存扛不住。
标签设计上有个坑：endpoint千万别用request.path。带 ID 的 RESTful 路径（/api/user/12345）会让每个用户 ID 变成一个独立的标签值，几万个标签直接把监控系统打挂。用request.endpoint（Flask 的路由函数名）或者request.url_rule.rule（带参数占位符的模板，如/api/user/<int:uid>），基数就固定在路由数量级别了。
5.2 从日志里挖耗时数据的两条路子
如果不想引入 Prometheus（有些团队就是不想多维护一套），纯日志也能做出可用的分析。核心是把耗时数据变成能被快速过滤的形式。
第一条路子是分级打标。不要只写一个duration_ms=234.5，同时写一个粗粒度的档位字段dur_bucket=200-500。这样在日志系统里直接按档位过滤，不需要做数值范围查询，速度快得多，尤其在没有索引的日志系统里。
def _bucket(ms): for edge in (50, 100, 200, 500, 1000, 3000, 10000): if ms < edge: return f"lt{edge}" return "gte10000"第二条路子是慢请求单独打一条。正常请求一条 INFO，超过阈值的额外补一条 WARNING，带上更多上下文（比如调用的下游服务、SQL 条数）。这样告警规则可以直接挂在这条 WARNING 上，简单粗暴但极其有效。代价是同一个请求有两条日志，分析时要按request_id去重。
这条 WARNING 里该放什么，我的经验是放排查入口而不是排查结论：不要写"这个请求很慢"，要写"这个请求慢了 2340ms，其中数据库相关耗时 1800ms，缓存命中 0 次，下游接口调用 3 次"。这些东西在请求处理的最后阶段才汇总得出来，所以通常需要你在视图里或钩子里用g累计一些计数器。
g._db_time = 0.0 g._cache_hits = 0 g._downstream_calls = 0配合 SQLAlchemy 的事件钩子，把每条 SQL 的执行时间累加到g._db_time上，这个数据的价值很高——大部分"接口慢"的问题，真相都是 N+1 查询，而这个计数器能一秒定位。
5.3 采样和降噪，别让日志把盘撑爆
一个日请求量千万级的服务，全量记录结构化日志，每天产生的数据量是按 GB 算的。如果每条日志还带完整的 UA 和一堆自定义字段，磁盘和采集带宽都吃不消。
分级策略我一般这么配：
- 错误（4xx 中的 401/403/429、全部 5xx）：100% 全量记录，这类是排查的核心。
- 慢请求（超过该接口 P95 阈值）：100% 全量记录，量本来就小。
- 正常请求：按比例采样，比如 10%，用一个稳定的哈希（对
request_id取模）保证同一个请求的多次记录采到或采不到是一致的。
采样率别用随机数。用request_id的哈希值取模，好处是同一个请求在所有环节的采样决策一致，不会出现"入口日志有、出口日志没有"的断裂。而且采样率调整之后，已经落盘的样本还能按request_id反查完整链路。
def _sampled(request_id, rate=0.1): return int(request_id[:8], 16) % 1000 < rate * 1000uuid4().hex的前 8 位是 32 bit，用它做取模足够了。注意如果request_id是从上游透传的、格式是自定义的短字符串，int(..., 16)可能抛异常，得用hashlib.md5兜底。
6. 上线之后才发现的七个细节
前面讲的都是能提前设计好的部分，这一节是在真实流量下撞出来的问题，都属于"文档不会写、不踩不知道"的类型。
反向代理的排队时间不在这里面。用户感知的等待时间，往往有 30% 花在网关排队、连接池等待这些环节，而你的duration_ms完全看不到。解决办法是让代理层把请求到达时间透传下来，比如通过一个X-Request-Start头传毫秒时间戳，应用层算差值得出总耗时，跟自己的处理耗时并列记录。两者的差值就是"框架之外的开销"，这个指标在容量规划时非常关键。
响应压缩的时间不在计时范围内。gzip 压缩通常由 WSGI 中间件或者代理层完成，如果它跑在你的计时中间件之后，压缩耗时就不会被记录。一个返回 2MB JSON 的接口，压缩可能就要几十毫秒。要么把计时放在压缩之后（其实做不到，因为压缩还在生成响应体），要么接受这个偏差并在文档里说明。
日志默认时间是本地时间，容器里通常是 UTC。%(asctime)s用的是time.localtime，容器一般设的是 UTC。等你跟运维核对"这个问题发生在下午三点"的时候，就会发现对方看的是北京时间，日志里是 UTC，差 8 小时。统一用 ISO8601 带时区偏移的格式（比如在 formatter 里显式指定%z），或者干脆全公司统一 UTC，别在这种地方埋雷。
日志里记请求体会让内存翻倍。request.get_data()会把请求体整个读到内存，如果是个 50MB 的上传请求，内存瞬间涨 50MB。而且 Flask 会缓存这个结果，后续代码再读不会重复消耗流，但内存已经占住了。要记录请求体的话，先判断request.content_length，超过几 KB 就只记长度和摘要哈希，别记原文。
teardown_request里数据库 session 可能已经关了。如果你用flask_sqlalchemy，session 通常也在teardown_appcontext里被移除。而teardown_request的执行顺序在 appcontext 之前，理论上还能用，但如果你的项目里有自定义的清理逻辑把这个顺序打乱了，就会拿到一个已关闭的 session。在 teardown 里补日志的时候，别做任何数据库查询，只用g里已经缓存好的数据。
生产环境的 Werkzeug 日志会干扰你的格式。开发服务器每个请求打一行127.0.0.1 - - [time] "GET / HTTP/1.1" 200 -，这行来自werkzeuglogger，跟你配的格式不一致。前面 dictConfig 里把werkzeug的 level 设成WARNING就能压掉。但如果用的是 gunicorn，访问日志是它的access_log_format控制的，那是另一套配置——%(D)s是微秒级的请求耗时，%(L)s是秒，别搞混了单位，我见过因为这个把 3 秒当成 3 毫秒报上去的。
时间单位不统一是最隐蔽的错误。perf_counter()差值是秒，time.time()差值也是秒，但日志里经常有人写duration_ms却塞进去秒值。看板上显示"平均耗时 0.23ms"，全组欢天喜地，实际是 230ms。我的做法是在变量名里强制带单位后缀，_t0_s、duration_ms、duration_us，看一眼就知道。聚合的时候再加一层断言，if duration_ms > 600000: 打警告，超过十分钟的单个请求基本可以断定是单位错了或者时钟出了问题。
最后分享一个我用得最顺手的小技巧：在日志格式里加一个进程号和线程名的短标识，%(process)d和%(threadName)s。排查并发问题时，你会需要知道"这两条日志是不是同一个 worker 打的"，多进程环境下光靠request_id关联是不够的——比如你怀疑某个 worker 的数据库连接池被打满了，没有进程号就只能干瞪眼。这个字段成本几乎为零，加上去不亏。
的排查与解决方法)
