Flask 的日志怎么配置?为什么线上看不到日志?
简化版
Flask 的 app.logger 就是一个标准库的 logging.Logger(名字是应用的 import_name),没有任何魔法——所以理解 Flask 日志的关键是理解 Python logging 的三层结构:Logger(产生日志)→ Handler(决定输出到哪)→ Formatter(决定长什么样),再加上两处独立的级别过滤(logger 的和 handler 的,两个都要够低才会输出)。「线上看不到日志」通常是四个原因:① 级别没配对——Flask 在非 debug 模式下 app.logger 的有效级别是 WARNING,logger.info() 直接被丢弃;② 配置时机太晚——一旦访问过 app.logger,Flask 就会给它加上默认 handler,之后再配 dictConfig 可能重复或冲突,所以配置必须在创建 app 之前或紧随其后;③ 写到了文件但容器里没人看——容器化部署应该直接输出到 stdout/stderr,让日志收集器去采,写文件在容器重启后就丢了;④ gunicorn 的 access log 和应用日志是两套——gunicorn.error/gunicorn.access 是独立的 logger,要单独配。推荐做法是用 logging.config.dictConfig 在 create_app() 之前统一配置(一处定义 formatter、handler、各 logger 的级别),生产环境输出 JSON 结构化日志到 stdout,并在每条日志里带上 request_id(用 before_request 生成 + logging.Filter 注入),这样一次请求的所有日志能被串起来。还有两条红线:日志里绝不能记录密码、token、完整卡号(最常见的泄露点是把整个 request.form 或请求头打进日志);用 logger.exception() 而不是 logger.error(str(e)),前者会带上完整堆栈。核心记忆:app.logger 就是标准 logging;级别有两处、都要够低;配置要早;容器里输出到 stdout;带 request_id。
详细版
logging 三层结构与常见配置:
| 组件 | 职责 | 关键点 |
|---|---|---|
| Logger | 产生日志、按名字分层 | 有自己的 level,默认向上传播 |
| Handler | 输出到哪(控制台/文件/网络) | 也有自己的 level |
| Formatter | 日志长什么样 | 时间、级别、模块、消息 |
| Filter | 过滤或注入字段 | 注入 request_id 靠它 |
propagate | 是否向父 logger 传递 | 不关会重复输出 |
# ① ★★推荐:dictConfig 在 create_app 之前配置★★
import logging.config, os, sys
def configure_logging():
logging.config.dictConfig({
"version": 1,
"disable_existing_loggers": False, # ★★不要禁用已有 logger★★
"filters": {
"request_id": {"()": "myapp.logging_ext.RequestIdFilter"},
},
"formatters": {
"default": {
"format": "[%(asctime)s] %(levelname)s in %(module)s "
"[%(request_id)s]: %(message)s",
},
"json": {"()": "pythonjsonlogger.jsonlogger.JsonFormatter",
"format": "%(asctime)s %(levelname)s %(name)s "
"%(message)s %(request_id)s"},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout", # ★★容器里输出到 stdout★★
"formatter": "json" if os.getenv("ENV") == "prod" else "default",
"filters": ["request_id"],
},
},
"root": {"level": "INFO", "handlers": ["console"]},
"loggers": {
"myapp": {"level": "DEBUG", "propagate": True},
"sqlalchemy.engine": {"level": "WARNING"}, # ★★别让 SQL 刷屏★★
"werkzeug": {"level": "WARNING"},
},
})
configure_logging() # ★★必须在 create_app() 之前★★
app = create_app()
# ② ★request_id 注入(把一次请求的日志串起来)★
import uuid, logging
from flask import g, has_request_context, request
class RequestIdFilter(logging.Filter):
def filter(self, record):
record.request_id = g.get("request_id", "-") if has_request_context() else "-"
record.path = request.path if has_request_context() else "-"
return True # ★★必须返回 True★★
@app.before_request
def _set_request_id():
g.request_id = request.headers.get("X-Request-Id") or uuid.uuid4().hex[:16]
@app.after_request
def _echo_request_id(resp):
resp.headers["X-Request-Id"] = g.get("request_id", "") # ★回显给客户端★
return resp
# ③ ★使用★
app.logger.info("用户登录 user_id=%s", user.id) # ★★用 %s 惰性格式化★★
app.logger.warning("库存不足 sku=%s left=%d", sku, n)
try:
do_something()
except Exception:
app.logger.exception("处理失败 order=%s", oid) # ★★带完整堆栈★★
# ✗ app.logger.error(str(e)) # ★丢了堆栈★
# 模块级 logger(★推荐★)
logger = logging.getLogger(__name__) # ★★自动带模块层级★★
# ④ ★gunicorn 的日志(★两套体系★)★
# gunicorn -w 4 --access-logfile - --error-logfile - \
# --access-logformat '%(h)s %(r)s %(s)s %(D)sµs "%({X-Request-Id}i)s"' \
# "myapp:create_app()"
# ★把 gunicorn 日志并入应用配置★
gunicorn_logger = logging.getLogger("gunicorn.error")
app.logger.handlers = gunicorn_logger.handlers
app.logger.setLevel(gunicorn_logger.level)
# ⑤ ★敏感信息脱敏★
SENSITIVE = {"password", "token", "secret", "authorization", "card_no"}
def safe_dict(d):
return {k: ("***" if k.lower() in SENSITIVE else v) for k, v in d.items()}
app.logger.info("form=%s", safe_dict(request.form.to_dict())) # ★★别直接打 form★★
⚠️ 三个必须记住的点:① 级别过滤有两处,两个都要够低才会输出:
logger.setLevel()和handler.setLevel()。一条INFO日志要经过「logger 的级别 → handler 的级别」两道关卡,任何一道设成WARNING它就消失了。而且 Flask 在非 debug 模式下app.logger的有效级别是WARNING(它没有显式设级别,会继承 root 的默认值WARNING)——这就是「本地logger.info看得见、线上什么都没有」的头号原因。② 配置日志必须尽早,最好在创建 app 之前。Flask 的app.logger是惰性创建的:第一次访问它时,如果对应的 logger 还没有 handler,Flask 会加上一个默认的StreamHandler。如果你在create_app()里先记了几条日志、之后才调dictConfig,就可能出现重复输出(默认 handler 和你配的都在)或配置被覆盖。所以标准做法是在create_app()调用之前执行configure_logging()。③ 容器化部署要输出到 stdout/stderr,不要写文件。写文件在容器里有三个问题:容器重启或重建就丢了、多副本时日志散落在各个容器里、磁盘写满会拖垮容器。十二要素应用的原则是「日志是事件流,应用只管往 stdout 写,收集和存储交给外部」(Docker logging driver、Fluentd、Loki)。真要写文件也必须用RotatingFileHandler/TimedRotatingFileHandler做轮转,而且多进程写同一个文件会互相覆盖——需要ConcurrentLogHandler或让每个进程写各自的文件。
完整版教学
一、logging 的三层结构
★ 一条日志的完整旅程:
logger.info("消息")
↓
① ★logger 级别检查★:logger.level > INFO? → ★丢弃★
↓
② logger 自己的 Filter
↓
③ ★遍历 logger.handlers★
↓
④ ★handler 级别检查★:handler.level > INFO? → ★该 handler 跳过★
↓
⑤ handler 的 Filter
↓
⑥ ★Formatter 格式化★
↓
⑦ 输出(控制台/文件/网络)
↓
⑧ ★propagate=True 则交给父 logger 重复上面的 ③~⑦★
↓
root logger
★ ★★两处级别是最常见的困惑★★:
logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG) # ★logger 放行 DEBUG★
h = logging.StreamHandler()
h.setLevel(logging.WARNING) # ★★但 handler 只收 WARNING★★
logger.addHandler(h)
logger.info("看不见") # ★★被 handler 挡住★★
★ ★规则:两道关卡都要过★
★ ★logger 的层级与 propagate★:
logging.getLogger("myapp") # 父
logging.getLogger("myapp.views") # ★子(按 . 分层)★
logging.getLogger("myapp.views.auth") # 孙
★ 子 logger 没设 level 时★继承最近的有设置的祖先★
★ propagate=True(默认)→ ★日志会一路交给父 logger 的 handler★
★ ★重复输出的经典原因★:
子 logger 加了 handler + propagate 没关 + root 也有 handler
→ ★同一条日志打印两次★
✓ 方案:★只在 root 上加 handler★,子 logger 只设 level
★ ★为什么推荐 logging.getLogger(__name__)★:
# myapp/views/auth.py
logger = logging.getLogger(__name__) # ★"myapp.views.auth"★
★ 好处:
① ★日志里能看出来自哪个模块★
② ★可以按模块调级别★(只把 myapp.views 调成 DEBUG)
③ ★天然的层级结构★
★ app.logger 的名字是 ★app.import_name★(通常是包名)
★ ★根 logger 的坑★:
logging.info("x") # ★直接用模块函数 = 用 root logger★
→ 如果 root 没有 handler,Python 会用 ★lastResort handler(只输出 WARNING+)★
→ ★所以 logging.info() 常常"什么都没发生"★
✓ ★永远用具名 logger★
★ ★basicConfig 的限制★:
logging.basicConfig(level=logging.INFO)
★ ✗ ★只在 root 没有 handler 时才生效★(第二次调用什么也不做)
★ ✗ 配置能力弱(一个 handler、一个 format)
✓ ★用 dictConfig★(可重复调用、能配多 handler/logger/filter)
理解 logging 要抓住一条日志的完整旅程:logger 级别检查 → logger 的 filter → 遍历 handlers → handler 级别检查 → handler 的 filter → formatter → 输出 → 按 propagate 交给父 logger。两处级别是最常见的困惑——logger 设了 DEBUG 但 handler 设了 WARNING,info() 照样看不见,两道关卡都要过。logger 按 . 分层,子 logger 没设 level 时继承最近的有设置的祖先;重复输出的经典原因是「子 logger 加了 handler + propagate 没关 + root 也有 handler」,解法是只在 root 上加 handler、子 logger 只设 level。推荐用 logging.getLogger(__name__)——日志里能看出模块来源、可以按模块调级别。还有个坑:直接用 logging.info() 是在用 root logger,root 没配 handler 时 Python 只会用 lastResort handler(只输出 WARNING 以上),所以「什么都没发生」。
二、Flask 的日志特性
★ ★app.logger 是什么★:
app.logger → logging.getLogger(app.name)
# app.name 通常是你的包名,如 "myapp"
★ ★它就是一个普通的 Logger,没有任何魔法★
★ ★Flask 的默认行为(★要理解才不会踩坑★)★:
Flask 在 ★第一次访问 app.logger 时★:
① 如果这个 logger ★还没有 handler★
② 且 ★没有设置 level★
→ ★加上一个默认的 StreamHandler(输出到 stderr)★
→ ★level 保持 NOTSET(继承 root 的 WARNING)★
★ 后果:
app.logger.info("x") # ★★非 debug 模式下看不见★★
app.logger.warning("y") # ✓ 能看见
★ ★这就是"线上 info 日志消失"的头号原因★
★ ★debug 模式的差别★:
app.debug = True → ★app.logger 的 level 被设为 DEBUG★
→ 本地开发时 info/debug 都能看到
→ ★上生产(debug=False)后突然全没了★
★ ★"本地有线上没有"的经典场景★
★ ★正确的配置时机★:
✗ def create_app():
app = Flask(__name__)
app.logger.info("starting") # ★★这里就触发了默认 handler★★
logging.config.dictConfig(...) # ★太晚了★
return app
✓ configure_logging() # ★★① 先配置★★
app = create_app() # ② 再创建
✓ 或者在 create_app 最开头:
def create_app():
logging.config.dictConfig(...) # ★在任何日志调用之前★
app = Flask(__name__)
...
★ ★Flask 相关的 logger 名字★:
┌────────────────────┬──────────────────────────────┐
│ ★app.name(你的包)★│ app.logger │
│ ★werkzeug★ │ ★开发服务器的请求日志★ │
│ ★sqlalchemy.engine★ │ ★SQL 语句(★调 INFO 会刷屏★)★ │
│ gunicorn.error │ ★gunicorn 自身的日志★ │
│ gunicorn.access │ ★访问日志★ │
│ celery / kombu │ Celery 相关 │
└────────────────────┴──────────────────────────────┘
★ 生产建议:
werkzeug → WARNING(生产不用它)
sqlalchemy.engine → ★WARNING★(★INFO 会打印每一条 SQL★)
你的应用 → INFO
★ ★app.logger vs 模块 logger★:
# ✓ 推荐:模块级
logger = logging.getLogger(__name__)
# ✓ 也可以:在视图里用 current_app.logger
from flask import current_app
current_app.logger.info(...)
★ ★两者最终都走同一套 handler★(只要配置得当)
★ 模块 logger 的优势:★不依赖应用上下文★(工具函数、后台任务里也能用)
★ ★错误日志:Flask 自动记什么★:
未处理异常 → ★Flask 自动 app.logger.error("Exception on ...", exc_info=True)★
→ ★所以不用自己在 errorhandler 里重复记★(除非要加业务字段)
★ 但 ★abort(404) 这类 HTTPException 不会被记录★
app.logger 就是 logging.getLogger(app.name),没有任何魔法。但 Flask 有个要理解才不会踩坑的默认行为:第一次访问 app.logger 时,如果它还没有 handler,Flask 会加一个默认的 StreamHandler,而 level 保持 NOTSET(继承 root 的 WARNING)——所以非 debug 模式下 app.logger.info() 看不见,这是「线上 info 日志消失」的头号原因;而 debug 模式会把 level 设为 DEBUG,于是「本地能看见、上线全没了」。配置时机因此很关键:必须在任何日志调用之前完成,推荐 configure_logging() 在 create_app() 之前执行。生产环境要单独调低几个 logger 的级别:werkzeug 设 WARNING、sqlalchemy.engine 设 WARNING(设 INFO 会打印每一条 SQL)。最后一个实用信息:未处理异常 Flask 会自动用 exc_info=True 记录,不用在 errorhandler 里重复记;但 abort(404) 这类 HTTPException 不会被记录。
三、结构化日志与 request_id
★ ★为什么要结构化(JSON)日志★:
★传统文本★:
[2026-08-02 10:00:00] INFO in views: 用户 42 下单成功,金额 99.5
→ ★要查"金额大于 100 的订单"只能写正则★
★结构化★:
{"ts":"2026-08-02T10:00:00Z","level":"INFO","logger":"myapp.views",
"msg":"order created","user_id":42,"amount":99.5,
"request_id":"a1b2c3","path":"/api/orders","duration_ms":45}
→ ★ELK/Loki 里直接按字段查询和聚合★
★ ★实现(python-json-logger)★:
pip install python-json-logger
"formatters": {
"json": {
"()": "pythonjsonlogger.jsonlogger.JsonFormatter",
"format": "%(asctime)s %(levelname)s %(name)s %(message)s "
"%(request_id)s %(module)s %(lineno)d",
}
}
# 记录额外字段
logger.info("order created", extra={"user_id": 42, "amount": 99.5})
★ ★注意:extra 里的 key 不能和 LogRecord 的内置属性冲突★
(message / args / exc_info / levelname 等会报错)
★ ★★request_id:把一次请求的日志串起来(最有价值的实践)★★:
没有它:
[10:00:00] INFO 开始处理
[10:00:00] INFO 查询用户 ← ★★这三条是同一个请求的吗?★★
[10:00:00] ERROR 支付失败 ← ★高并发下日志交错★
有了它:
[10:00:00] INFO [a1b2c3] 开始处理
[10:00:00] INFO [a1b2c3] 查询用户
[10:00:00] ERROR [a1b2c3] 支付失败 ← ★★grep a1b2c3 就是完整链路★★
★ 实现三步:
① before_request 生成或透传
g.request_id = request.headers.get("X-Request-Id") or uuid4().hex[:16]
② ★logging.Filter 注入到每条日志★
class RequestIdFilter(logging.Filter):
def filter(self, record):
record.request_id = (g.get("request_id", "-")
if has_request_context() else "-")
return True # ★★必须返回 True 否则日志被丢弃★★
③ after_request 回显给客户端
resp.headers["X-Request-Id"] = g.request_id
★ ★价值:用户报障时提供这个 ID,你能立刻定位到完整的处理过程★
★ ★跨服务传递★:
调用下游服务时把 X-Request-Id 带上
→ ★整条调用链都能串起来★(这就是分布式追踪的雏形)
★ ★异步任务里★:
投递 Celery 任务时把 request_id 作为参数传过去
→ 任务日志也能关联回原始请求
★ ★请求访问日志(自己实现)★:
@app.before_request
def _start(): g._t0 = time.perf_counter()
@app.after_request
def _access_log(resp):
logger.info("request", extra={
"method": request.method, "path": request.path,
"status": resp.status_code,
"duration_ms": round((time.perf_counter() - g._t0) * 1000, 1),
"ip": request.remote_addr, # ★要先配 ProxyFix★
"ua": request.headers.get("User-Agent", "")[:200],
"user_id": getattr(current_user, "id", None),
})
return resp
★ ★注意:after_request 在未处理异常时不执行★
→ 需要完整覆盖就用 teardown_request
★ ★该记什么级别★:
┌──────────┬────────────────────────────────────────┐
│ DEBUG │ ★开发排查用★,生产一般关闭 │
│ ★INFO★ │ ★正常的业务事件★(登录、下单、任务完成) │
│ ★WARNING★ │ ★可恢复的异常★(重试成功、降级、参数异常)│
│ ★ERROR★ │ ★需要人关注的失败★(异常、外部服务不可用)│
│ CRITICAL │ ★服务不可用★ │
└──────────┴────────────────────────────────────────┘
★ ★原则:ERROR 应该是"需要有人看一眼"的★
→ ★如果每天几万条 ERROR,那它就失去意义了★
结构化(JSON)日志的价值是「可查询、可聚合」——传统文本日志要查「金额大于 100 的订单」只能写正则,而 JSON 日志在 ELK/Loki 里直接按字段查。request_id 是最有价值的日志实践:高并发下多个请求的日志会交错在一起,加上 request_id 后 grep 一下就是完整的处理链路;实现三步是「before_request 生成或透传 → logging.Filter 注入每条日志(filter() 必须返回 True,否则日志会被丢弃)→ after_request 回显给客户端」。它的实际价值是用户报障时提供这个 ID,你能立刻定位完整过程;调用下游服务时带上它,整条链路就能串起来(分布式追踪的雏形),投递 Celery 任务时也应该传过去。级别的选择原则是「ERROR 应该是需要有人看一眼的」——如果每天几万条 ERROR,它就失去意义了。
四、部署环境下的日志
★ ★★容器化:输出到 stdout,不要写文件★★:
十二要素应用原则:★"日志是事件流,应用只管往 stdout 写"★
★ 写文件在容器里的三个问题:
① ★容器重启/重建就丢了★
② ★多副本时日志散落在各容器★
③ ★磁盘写满会拖垮容器★
✓ "handlers": {"console": {"class": "logging.StreamHandler",
"stream": "ext://sys.stdout"}}
→ Docker logging driver / Fluentd / Loki / CloudWatch 负责收集
★ ★传统部署:文件 + 轮转★:
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
RotatingFileHandler("app.log", maxBytes=50*1024*1024, backupCount=10)
TimedRotatingFileHandler("app.log", when="midnight", backupCount=30)
★ ★★多进程写同一个文件的问题★★:
gunicorn 8 个 worker 同时写 app.log
→ ★轮转时互相覆盖★(一个进程 rename 了,其他还在写旧 fd)
→ ★可能丢日志或产生 .log.1 混乱★
✓ 方案一:★每个进程写各自的文件★(app-{pid}.log)
✓ 方案二:★ConcurrentLogHandler(第三方,带文件锁)★
✓ 方案三:★★交给 logrotate(外部轮转 + copytruncate)★★
✓ 方案四:★★根本不写文件,输出到 stdout 由 supervisor/systemd 接管★★
★ ★gunicorn 的两套日志★:
┌──────────────────┬────────────────────────────────┐
│ ★gunicorn.error★ │ ★启动、worker 生命周期、异常★ │
│ ★gunicorn.access★ │ ★每个请求的访问日志★ │
└──────────────────┴────────────────────────────────┘
# 命令行
gunicorn --access-logfile - --error-logfile - \
--access-logformat '%(h)s "%(r)s" %(s)s %(b)s %(D)s "%({X-Request-Id}i)s"'
★ ★- 表示输出到标准输出★
★ 常用变量:%(h)s IP、%(r)s 请求行、%(s)s 状态码、
★%(D)s 耗时(微秒)★、★%({Header}i)s 请求头★、%({Header}o)s 响应头
★ ★把 app.logger 并入 gunicorn 的 handler★:
gunicorn_logger = logging.getLogger("gunicorn.error")
app.logger.handlers = gunicorn_logger.handlers
app.logger.setLevel(gunicorn_logger.level)
★ 好处:★统一输出目标和格式★
★ 或者反过来:用 --logger-class 或 dictConfig 统一接管
★ ★uWSGI★:
uwsgi --logto /var/log/app.log 或 --log-master
★ uWSGI 会★劫持 stdout★,Python logging 的行为可能不符预期
✓ 用 ★--log-format★ 或让 Python 日志走 stderr
★ ★Nginx 层的日志★:
log_format main '$remote_addr "$request" $status $body_bytes_sent '
'$request_time "$http_x_request_id"'; # ★★传递 request_id★★
★ ★三层日志要能对上★:Nginx → gunicorn access → 应用日志
→ ★靠 request_id 串联★
★ ★日志量控制(★成本和性能★)★:
① ★别在循环里打日志★(一次请求几千条)
② ★用 %s 惰性格式化★:
✓ logger.info("user %s", uid) # ★级别不够时不做格式化★
✗ logger.info(f"user {uid}") # ★★f-string 总是先算出来★★
✗ logger.info("user " + str(uid))
③ ★采样★:高频日志只记 1%
if random.random() < 0.01: logger.info(...)
④ ★大对象不要整个打★(截断或只打关键字段)
⑤ ★sqlalchemy.engine 保持 WARNING★
★ ★异步日志(高吞吐场景)★:
from logging.handlers import QueueHandler, QueueListener
q = queue.Queue(-1)
listener = QueueListener(q, console_handler)
listener.start()
# logger 只挂 QueueHandler → ★写日志不阻塞请求线程★
★ 适合:★handler 慢时★(写网络、写慢磁盘)
容器化部署的原则是「日志是事件流,应用只管往 stdout 写」——写文件在容器里会丢、会散落、会写满磁盘。传统部署写文件必须做轮转,但要注意多进程写同一个文件的问题:8 个 worker 同时写,轮转时会互相覆盖(一个进程 rename 了,其他还在写旧 fd);四种解法是每进程独立文件、ConcurrentLogHandler、交给 logrotate 用 copytruncate、或干脆输出到 stdout 由 supervisor 接管。gunicorn 有两套独立的 logger(gunicorn.error 和 gunicorn.access),可以把 app.logger.handlers 指向 gunicorn 的来统一输出。三层日志(Nginx → gunicorn access → 应用)要靠 request_id 串联。日志量控制里有个必须记住的细节:用 logger.info("user %s", uid) 而不是 f-string——前者在级别不够时不会做格式化,后者总是先算出来。
五、日志安全与合规
★ ★★绝不能记录的内容★★:
✗ ★密码★(哪怕是错误的密码)
✗ ★token / API Key / SECRET_KEY★
✗ ★完整的银行卡号、身份证号★
✗ ★Cookie / Authorization 头★
✗ ★用户的隐私内容★(私信、健康数据)
★ ★最常见的泄露点(★都很隐蔽★)★:
① ★把整个表单打进日志★
logger.info("form=%s", request.form.to_dict()) # ★★含密码★★
② ★把请求头全打★
logger.debug("headers=%s", dict(request.headers)) # ★★含 Authorization/Cookie★★
③ ★异常堆栈里的局部变量★
某些日志库(如 better-exceptions)会打印局部变量 → ★可能含密码★
④ ★SQL 日志里的参数★
sqlalchemy.engine 的 INFO 级别会打印 ★带参数的完整 SQL★
⑤ ★第三方库的 debug 日志★
requests/urllib3 的 DEBUG 会打印 ★完整请求头(含 token)★
⑥ ★URL 里的敏感参数★
/reset?token=xxx → ★access log 里就有了★
✓ ★敏感参数放 POST body 而不是 query string★
★ ★脱敏实现★:
SENSITIVE_KEYS = {"password", "passwd", "pwd", "token", "secret",
"authorization", "cookie", "api_key", "card_no"}
def mask(d):
return {k: ("***" if any(s in k.lower() for s in SENSITIVE_KEYS) else v)
for k, v in d.items()}
# ★更彻底:用 Filter 全局脱敏★
import re
PATTERNS = [
(re.compile(r'("password"\s*:\s*)"[^"]*"'), r'\1"***"'),
(re.compile(r'(Bearer\s+)[\w\-.]+'), r'\1***'),
(re.compile(r'\b\d{16,19}\b'), "****"), # ★卡号★
]
class MaskFilter(logging.Filter):
def filter(self, record):
msg = record.getMessage()
for pat, rep in PATTERNS:
msg = pat.sub(rep, msg)
record.msg, record.args = msg, ()
return True
★ ★注意:正则脱敏是兜底,不能替代"一开始就不记"★
★ ★合规相关★:
□ ★日志保留期★(GDPR 要求不能无限期保留个人数据)
□ ★访问控制★(谁能看生产日志)
□ ★审计日志单独存★(登录、权限变更、敏感操作)
□ ★日志本身也是审计对象★(谁在什么时候查了日志)
★ ★审计日志(和普通日志分开)★:
audit_logger = logging.getLogger("audit")
# 单独的 handler、更长的保留期、不可删除的存储
audit_logger.info("permission_changed", extra={
"actor_id": current_user.id, "target_id": u.id,
"before": old_role, "after": new_role,
"ip": request.remote_addr, "request_id": g.request_id,
})
★ 要点:★谁、对谁、做了什么、什么时候、从哪★
★ ★日志告警★:
# Sentry:自动捕获 ERROR 及以上
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration
sentry_sdk.init(dsn=..., integrations=[
LoggingIntegration(level=logging.INFO, # ★面包屑级别★
event_level=logging.ERROR) # ★★上报级别★★
])
★ ★别把所有 ERROR 都告警★ → 告警疲劳
✓ 分级:★P0 立即电话、P1 企微/钉钉、P2 每日汇总★
日志泄露的六个常见点都很隐蔽:把整个 request.form 打进日志(含密码)、把请求头全打(含 Authorization 和 Cookie)、异常堆栈里的局部变量、sqlalchemy.engine 的 INFO 级别会打印带参数的完整 SQL、requests/urllib3 的 DEBUG 会打印完整请求头、以及 URL 里的敏感参数会进 access log(所以敏感参数要放 POST body 而不是 query string)。脱敏可以用全局的 logging.Filter 做正则替换,但正则脱敏只是兜底,不能替代「一开始就不记」。审计日志要和普通日志分开(单独 handler、更长保留期、不可删除的存储),记录「谁、对谁、做了什么、什么时候、从哪」。告警要分级——别把所有 ERROR 都告警,那会造成告警疲劳。
六、实践清单
★ ★完整的日志模块模板★:
# myapp/logging_ext.py
import logging, uuid, os, sys
from flask import g, request, has_request_context
class RequestIdFilter(logging.Filter):
def filter(self, record):
if has_request_context():
record.request_id = g.get("request_id", "-")
record.path = request.path
record.method = request.method
else:
record.request_id = record.path = record.method = "-"
return True
def configure_logging():
is_prod = os.getenv("ENV") == "production"
logging.config.dictConfig({
"version": 1, "disable_existing_loggers": False,
"filters": {"rid": {"()": RequestIdFilter}},
"formatters": {
"text": {"format": "[%(asctime)s] %(levelname)s "
"%(name)s [%(request_id)s] %(message)s"},
"json": {"()": "pythonjsonlogger.jsonlogger.JsonFormatter",
"format": "%(asctime)s %(levelname)s %(name)s "
"%(message)s %(request_id)s %(path)s"},
},
"handlers": {"console": {
"class": "logging.StreamHandler", "stream": "ext://sys.stdout",
"formatter": "json" if is_prod else "text", "filters": ["rid"]}},
"root": {"level": "INFO", "handlers": ["console"]},
"loggers": {
"myapp": {"level": "INFO" if is_prod else "DEBUG"},
"werkzeug": {"level": "WARNING"},
"sqlalchemy.engine": {"level": "WARNING"},
},
})
def init_request_logging(app):
@app.before_request
def _s():
g.request_id = request.headers.get("X-Request-Id") or uuid.uuid4().hex[:16]
g._t0 = time.perf_counter()
@app.after_request
def _e(resp):
resp.headers["X-Request-Id"] = g.get("request_id", "")
logging.getLogger("myapp.access").info(
"request", extra={"status": resp.status_code,
"duration_ms": round((time.perf_counter()-g._t0)*1000, 1)})
return resp
★ 检查清单:
□ ★configure_logging() 在 create_app() 之前调用★
□ ★disable_existing_loggers: False★
□ ★生产用 JSON 格式输出到 stdout★
□ ★每条日志带 request_id★
□ ★request_id 回显给客户端并透传给下游★
□ ★sqlalchemy.engine / werkzeug 设 WARNING★
□ ★用 logger.exception 而不是 error(str(e))★
□ ★用 %s 惰性格式化而不是 f-string★
□ ★不记录密码/token/完整卡号★
□ ★不整个打印 request.form 和 headers★
□ ★ERROR 级别接告警(分级,避免疲劳)★
□ ★写文件必须轮转,多进程要处理并发★
★ ★排查"看不到日志"的顺序★:
① ★logger 的有效级别★:logging.getLogger("myapp").getEffectiveLevel()
② ★handler 的级别★:[h.level for h in logger.handlers]
③ ★有没有 handler★:logger.handlers(空的话看 propagate 和 root)
④ ★propagate 是否被关★
⑤ ★配置时机★(是不是在第一条日志之后才配的)
⑥ ★输出目标★(写到文件了?容器里 stdout 才能被采到)
⑦ ★gunicorn 是否接管了★
★ 一句话总结:
★"app.logger 就是标准 logging——级别有 logger 和 handler 两处、
都要够低;Flask 非 debug 下默认是 WARNING,所以 info 会消失;
配置要在创建 app 之前用 dictConfig 做完;
容器里输出 JSON 到 stdout,每条带 request_id 串起一次请求;
绝不记密码 token,用 %s 惰性格式化。"★
检查清单里最容易漏的四条:disable_existing_loggers: False(否则会禁用第三方库的 logger)、request_id 要透传给下游服务、用 %s 惰性格式化而不是 f-string、写文件必须处理多进程并发。排查「看不到日志」有固定顺序:先看 getEffectiveLevel()、再看各 handler 的 level、然后确认有没有 handler、propagate 是否被关、配置时机对不对、以及输出目标是不是能被采集到。
记忆钩子:「★app.logger 就是 logging.getLogger(app.name),没有任何魔法★——所以要理解 Python logging 的三层:★Logger(产生)→ Handler(输出到哪)→ Formatter(长什么样)★,外加 Filter(过滤或★注入字段★)。★最常见的困惑是级别有两处:logger 和 handler,两道关卡都要过★(logger 设 DEBUG 但 handler 设 WARNING,info 照样看不见)。★『线上看不到日志』的四个原因★:★① Flask 非 debug 模式下 app.logger 的有效级别是 WARNING★(它不显式设 level,继承 root 默认值),而★debug 模式会设成 DEBUG★ → ★这就是『本地 info 看得见、线上全没了』的头号原因★;★② 配置时机太晚★——Flask 在★第一次访问 app.logger 时★会给没有 handler 的 logger 加一个默认 StreamHandler,所以 ★configure_logging() 必须在 create_app() 之前★,否则重复输出或被覆盖;★③ 写到文件但容器里没人看★——★十二要素原则是『日志是事件流,应用只管往 stdout 写』★,写文件在容器里会丢、会散落、会写满磁盘;★④ gunicorn 的 gunicorn.error / gunicorn.access 是两套独立 logger★,要单独配或把 app.logger.handlers 指过去。★重复输出的经典原因:子 logger 加了 handler + propagate 没关 + root 也有 handler★ → 解法是★只在 root 加 handler、子 logger 只设 level★。推荐 ★logging.getLogger(name)★(带模块层级、可按模块调级别、不依赖应用上下文),★别直接用 logging.info()★(那是 root logger,没配 handler 时只输出 WARNING+)。生产要调低两个:★werkzeug → WARNING、sqlalchemy.engine → WARNING★(设 INFO 会打印每条 SQL)。★request_id 是最有价值的实践★:before_request 生成或透传 → ★logging.Filter 注入(filter() 必须 return True 否则日志被丢弃)★ → after_request 回显;★grep 一个 id 就是完整链路★,★调下游和投 Celery 时都要带上★。★用 logger.exception() 而不是 error(str(e))★(前者带完整堆栈),★用 %s 惰性格式化而不是 f-string★(级别不够时不做格式化)。★未处理异常 Flask 会自动 exc_info=True 记录,但 abort(404) 这类 HTTPException 不会★。安全红线:★绝不记密码/token/完整卡号★,最隐蔽的泄露点是★整个打印 request.form(含密码)和 headers(含 Authorization/Cookie)★,还有 ★URL 里的敏感参数会进 access log★(所以敏感参数放 POST body)。多进程写同一个文件★轮转时会互相覆盖★ → 用 logrotate 的 copytruncate 或每进程独立文件或干脆输出 stdout。」
七、常见误区与追问
- 误区:
app.logger.info()写了就一定能看到日志。 非 debug 模式下看不到。Flask 的app.logger没有显式设置级别(保持NOTSET),于是它继承 root logger 的默认级别WARNING——INFO和DEBUG都会被直接丢弃。而在app.debug = True时 Flask 会把它设成DEBUG,所以本地开发时 info 日志都正常,一部署到生产就全部消失,这是最经典的「本地有线上没有」场景。而且级别过滤有两处:logger.setLevel()和handler.setLevel()——两道关卡都要够低才会输出,很多人只改了其中一个还在纳闷。排查时打印logging.getLogger("myapp").getEffectiveLevel()和[h.level for h in logger.handlers]就能立刻定位。正确做法是用dictConfig显式声明每个 logger 和 handler 的级别,别依赖默认值。 - 误区:在
create_app()里随便什么位置配置日志都行。 必须在第一次访问app.logger之前。Flask 的app.logger是惰性创建的:第一次访问时,如果对应的 logger 还没有任何 handler,Flask 会自动加上一个默认的StreamHandler(输出到 stderr)。所以如果你在create_app()里先写了app.logger.info("starting..."),之后才调logging.config.dictConfig(...),结果就是同一条日志被打印两次(Flask 的默认 handler 和你配的 handler 都在),或者格式混乱。标准做法是把configure_logging()放在create_app()调用之前(在wsgi.py或app.py的模块顶部),或者放在create_app()函数的第一行、在创建Flask()实例之前。另外dictConfig里一定要写"disable_existing_loggers": False——默认是True,会把此前已创建的所有 logger(包括第三方库的)全部禁用掉。 - 误区:日志写到文件里最可靠,容器里也一样。 容器里写文件有三个问题:① 容器重启或重建,文件就没了——而排查问题恰恰经常发生在容器崩溃之后;② 多副本时日志散落在各个容器里,你得逐个
docker exec进去看;③ 磁盘写满会拖垮容器(容器的可写层通常很小)。十二要素应用的原则是「日志是事件流,应用只管往 stdout 写」——由 Docker logging driver、Fluentd、Loki、CloudWatch 这类外部设施负责收集、聚合、存储和轮转,应用完全不关心。传统的物理机/虚拟机部署可以写文件,但必须做轮转(RotatingFileHandler或TimedRotatingFileHandler),而且要处理一个隐蔽的问题:多个 gunicorn worker 同时写同一个文件时,轮转会互相覆盖(一个进程 rename 了文件,其他进程还在往旧的 fd 里写),可能丢日志——解法是让每个进程写各自的文件、用带文件锁的ConcurrentLogHandler、或者交给外部的 logrotate 用copytruncate模式。 - 误区:
logger.error(f"处理失败: {e}")已经把异常信息记下来了。 丢掉了最重要的堆栈。str(e)只有异常的消息文本(很多异常的消息还很含糊,比如KeyError: 'user_id'),你完全不知道它是在哪一行、经过什么调用路径抛出来的——排查时只能靠猜。正确的写法是logger.exception("处理失败 order=%s", order_id)——它等价于logger.error(..., exc_info=True),会自动把完整的 traceback 附在日志里(必须在except块内调用)。如果不在 except 块里但有异常对象,可以用logger.error("...", exc_info=e)。顺带说两个相关的点:Flask 对未处理的异常会自动调用app.logger.error(..., exc_info=True),所以不需要在errorhandler里重复记(除非要补充业务字段);而abort(404)这类HTTPException不会被自动记录,因为它们是正常的流程控制。 - 误区:用 f-string 写日志更简洁,和
%s没区别。 性能上有实际差别,而且差别在「日志不输出的时候」最明显。logger.debug(f"user {expensive_repr(obj)}")这行代码里,f-string 会立即求值——即使当前级别是INFO、这条 debug 日志根本不会被输出,字符串拼接和expensive_repr()的开销已经付出了。而logger.debug("user %s", obj)只是把参数存起来,只有确认这条日志真的要输出时才做格式化。在高频路径(每个请求几十条 debug 日志)上这个差别很可观。除了性能,%s风格还有两个好处:① 结构化日志库可以拿到原始的record.msg和record.args,从而把参数单独提取成字段;② Sentry 之类的工具可以按「消息模板」聚合同类错误("user %s not found"的一万次记录会被归为一个 issue,而 f-string 会产生一万个不同的消息)。所以规范是:日志一律用%s惰性格式化。 - 追问:
request_id具体怎么落地?为什么说它是最有价值的日志实践? 因为没有它,高并发下的日志基本没法读——同一时刻有几十个请求在处理,日志行交错在一起,你看到一条 ERROR 却不知道它之前发生了什么、是哪个用户触发的。落地分三步:① 在before_request里生成或透传——优先取请求头里的X-Request-Id(如果上游 Nginx 或网关已经生成了就复用),没有就uuid4().hex[:16],存进g;② 用logging.Filter注入到每一条日志记录——在filter()方法里给record加上request_id属性,然后在 formatter 的格式串里引用%(request_id)s;注意filter()必须return True,返回False意味着「丢弃这条日志」;还要用has_request_context()判断,因为 CLI 命令和后台任务里没有请求上下文。③ 在after_request里回显到响应头——这样用户报障时可以提供这个 ID,你grep一下就能看到那次请求的完整处理过程。进阶用法:调用下游服务时把它放进请求头传过去(整条调用链都能串起来,这就是分布式追踪的雏形)、投递 Celery 任务时作为参数传过去(异步任务的日志也能关联回原始请求)。 - 追问:gunicorn 的日志和 Flask 的日志是什么关系?怎么统一? 它们是完全独立的两套。gunicorn 有两个自己的 logger:
gunicorn.error(记录启动信息、worker 的生命周期事件、worker 超时被杀等)和gunicorn.access(每个请求的访问日志);而你的应用日志走的是app.logger或模块 logger。默认情况下它们的输出目标和格式都不一样,日志混在一起很难看。三种统一方式:① 把app.logger挂到 gunicorn 的 handler 上——app.logger.handlers = logging.getLogger("gunicorn.error").handlers,简单粗暴但会丢掉你自己的 formatter;② 反过来,用dictConfig统一接管所有 logger(包括gunicorn.error和gunicorn.access),这是最干净的做法——在配置里为它们指定和应用日志相同的 handler 和 formatter;③ 用 gunicorn 的--logger-class自定义。另外访问日志建议用--access-logformat加上耗时和 request_id:'%(h)s "%(r)s" %(s)s %(D)s "%({X-Request-Id}i)s"'——%(D)s是微秒级耗时,%({Header}i)s能取请求头。这样 Nginx 日志、gunicorn access 日志、应用日志三层就能靠 request_id 对上。 - 追问:日志里最容易泄露敏感信息的地方有哪些? 六个,而且都很隐蔽。① 把整个表单打进日志——
logger.info("form=%s", request.form.to_dict())在登录接口上就直接记下了明文密码;② 把请求头全打——dict(request.headers)里有Authorization(token)和Cookie(session);③sqlalchemy.engine的 INFO 级别——它会打印带绑定参数的完整 SQL,注册用户的那条 INSERT 里就有密码哈希甚至明文;④ 第三方库的 DEBUG 日志——requests/urllib3在 DEBUG 下会打印完整的请求头,调用第三方 API 的 API Key 就进日志了;⑤ 异常堆栈中的局部变量——某些增强型的异常格式化库会打印局部变量值,而登录函数的局部变量里就有密码;⑥ URL 里的敏感参数——/reset-password?token=xxx会出现在 Nginx access log、浏览器历史和 Referer 头里,所以敏感参数必须放 POST body 而不是 query string。防御是双层的:主要靠「一开始就不记」(脱敏字典、只记 id 不记内容),辅以全局的logging.Filter做正则兜底(替换掉形如Bearer xxx、"password": "..."、16~19 位连续数字的内容)——但要清楚正则脱敏只是最后一道防线,不能替代规范。
八、加强记忆
app.logger 就是 logging.getLogger(app.name),没有任何魔法——所以理解 Flask 日志的关键是理解 Python logging 的三层结构:Logger(产生)→ Handler(输出到哪)→ Formatter(长什么样),外加 Filter(过滤或注入字段)。最常见的困惑是级别有两处:logger 的和 handler 的,两道关卡都要过(logger 设了 DEBUG 但 handler 设了 WARNING,info() 照样看不见)。「线上看不到日志」有四个原因:① Flask 非 debug 模式下 app.logger 的有效级别是 WARNING(它不显式设 level,继承 root 的默认值),而 debug 模式会设成 DEBUG——这就是「本地 info 看得见、线上全没了」的头号原因;② 配置时机太晚——Flask 在第一次访问 app.logger 时会给没有 handler 的 logger 加一个默认 StreamHandler,所以 configure_logging() 必须在 create_app() 之前,否则会重复输出或配置被覆盖(还要记得写 disable_existing_loggers: False);③ 写到文件但容器里没人看——十二要素原则是「日志是事件流,应用只管往 stdout 写」,容器里写文件会丢、会散落、会写满磁盘;④ gunicorn 的 gunicorn.error 和 gunicorn.access 是两套独立的 logger,要单独配置或把 app.logger.handlers 指过去。重复输出的经典原因是「子 logger 加了 handler + propagate 没关 + root 也有 handler」——解法是只在 root 上加 handler、子 logger 只设 level。推荐用 logging.getLogger(__name__)(带模块层级、可按模块调级别、不依赖应用上下文),别直接用 logging.info()(那是 root logger,没配 handler 时只输出 WARNING 以上)。生产环境要调低两个:werkzeug 设 WARNING、sqlalchemy.engine 设 WARNING(设成 INFO 会打印每一条 SQL)。request_id 是最有价值的日志实践:before_request 生成或透传 → logging.Filter 注入(filter() 必须 return True,否则日志被丢弃) → after_request 回显;grep 一个 id 就是完整链路,调用下游和投递 Celery 时都要带上。写日志要用 logger.exception() 而不是 error(str(e))(前者带完整堆栈),用 %s 惰性格式化而不是 f-string(级别不够时不做格式化,还能让 Sentry 按模板聚合)。另外要知道未处理异常 Flask 会自动用 exc_info=True 记录,但 abort(404) 这类 HTTPException 不会。安全红线:绝不记录密码、token、完整卡号,最隐蔽的泄露点是整个打印 request.form(含密码)和 headers(含 Authorization/Cookie),还有 URL 里的敏感参数会进 access log(所以敏感参数要放 POST body)。最后,多进程写同一个文件时轮转会互相覆盖——用 logrotate 的 copytruncate、每进程独立文件、或干脆输出到 stdout。