FastAPI 的日志和可观测性怎么做？uvicorn 日志为什么不听话？

简化版

FastAPI 本身没有日志系统，用的就是标准库 logging——但它跑在 uvicorn 里，而 uvicorn 启动时会用自己的 LOGGING_CONFIG 配置 uvicorn、uvicorn.error、uvicorn.access 三个 logger，并且默认带 disable_existing_loggers: True——这就是「我明明配了 dictConfig，日志却不按我的格式输出」的根本原因：uvicorn 的配置在你之后执行，把你的 logger 禁用了。解法是在自己的 dictConfig 里一并接管 uvicorn 的三个 logger 并设 disable_existing_loggers: False，或者用 --log-config 传入自己的配置文件。第二个高频问题是「多 worker 下日志混在一起分不清」——解法是给每条日志带上 request_id：在中间件里生成（或从 X-Request-Id 透传），用 contextvars 存储（asyncio 下不能用 threading.local，协程之间会串），再用 logging.Filter 注入到每条记录。生产环境要输出 JSON 结构化日志到 stdout（容器里写文件会丢、会散落），并接入 APM。可观测性的三大支柱：日志（发生了什么）、指标（有多少、多快）、追踪（一次请求经过了哪些服务）——FastAPI 生态里 OpenTelemetry 的自动埋点最省事（一行 FastAPIInstrumentor.instrument_app(app) 就能拿到每个路由的耗时、SQL 语句、外部 HTTP 调用的完整链路）。还有三条红线：日志里绝不能出现密码和 token（最常见的泄露点是把整个请求体打进日志）、用 logger.exception() 而不是 logger.error(str(e))、用 %s 惰性格式化而不是 f-string。核心记忆：uvicorn 会覆盖你的日志配置；request_id 用 contextvars；容器输出 JSON 到 stdout；OTel 自动埋点。

详细版

日志配置的关键点：

项	要点
uvicorn 的 logger	`uvicorn`/`uvicorn.error`/`uvicorn.access`
`disable_existing_loggers`	必须设 `False`
配置时机	在 uvicorn 启动前或用 `--log-config`
request_id	contextvars（不能用 threading.local）
生产格式	JSON 到 stdout
追踪	OpenTelemetry 自动埋点

# ① ★★统一的日志配置（接管 uvicorn 的 logger）★★
import logging.config, os, sys

def setup_logging():
    is_prod = os.getenv("ENV") == "production"
    logging.config.dictConfig({
        "version": 1,
        ★"disable_existing_loggers": False★,          # ★★关键★★
        "filters": {"request_id": {"()": "myapp.logging_ext.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 %(pathname)s %(lineno)d"},
        },
        "handlers": {"console": {
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",             # ★★容器里输出到 stdout★★
            "formatter": "json" if is_prod else "text",
            "filters": ["request_id"],
        }},
        "root": {"level": "INFO", "handlers": ["console"]},
        "loggers": {
            "myapp": {"level": "DEBUG" if not is_prod else "INFO"},
            ★"uvicorn": {"handlers": ["console"], "level": "INFO",
                        "propagate": False}★,          # ★★接管 uvicorn★★
            ★"uvicorn.error": {"handlers": ["console"], "level": "INFO",
                              "propagate": False}★,
            ★"uvicorn.access": {"handlers": ["console"], "level": "WARNING",
                               "propagate": False}★,   # ★★自己记访问日志就关掉它★★
            "sqlalchemy.engine": {"level": "WARNING"},  # ★★INFO 会打印每条 SQL★★
        },
    })

setup_logging()                                        # ★★在 create_app 之前★★
app = create_app()

# ② ★★request_id：用 contextvars（不是 threading.local）★★
from contextvars import ContextVar
import uuid, logging

request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")

class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_ctx.get()
        return True                                    # ★★必须 return True★★

class RequestIdMiddleware:                             # ★纯 ASGI★
    def __init__(self, app): self.app = app
    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        headers = dict(scope["headers"])
        rid = (headers.get(b"x-request-id") or b"").decode() or uuid.uuid4().hex[:16]
        ★token = request_id_ctx.set(rid)★
        async def send_wrapper(message):
            if message["type"] == "http.response.start":
                message["headers"].append((b"x-request-id", rid.encode()))
            await send(message)
        try:
            await self.app(scope, receive, send_wrapper)
        finally:
            ★request_id_ctx.reset(token)★              # ★★清理★★

# ③ ★访问日志中间件（比 uvicorn.access 信息更全）★
class AccessLogMiddleware:
    def __init__(self, app): self.app = app
    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        t0 = time.perf_counter(); status = 500
        async def send_wrapper(message):
            nonlocal status
            if message["type"] == "http.response.start":
                status = message["status"]
            await send(message)
        try:
            await self.app(scope, receive, send_wrapper)
        finally:
            logging.getLogger("myapp.access").info("request", extra={
                "method": scope["method"], "path": scope["path"],
                "status": status,
                "duration_ms": round((time.perf_counter() - t0) * 1000, 1),
                "client": scope.get("client", ("-",))[0],
            })

# ④ ★使用★
logger = logging.getLogger(__name__)                   # ★★模块级★★
logger.info("order created order_id=%s amount=%s", oid, amt)  # ★★%s 惰性★★
try: ...
except Exception:
    logger.exception("处理失败 order_id=%s", oid)       # ★★带完整堆栈★★

# ⑤ ★OpenTelemetry 自动埋点★
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

FastAPIInstrumentor.instrument_app(app, excluded_urls="/health,/metrics")
SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine)
HTTPXClientInstrumentor().instrument()
# → ★自动记录：路由耗时、SQL 语句和耗时、外部 HTTP 调用★

# ⑥ ★Prometheus 指标★
from prometheus_fastapi_instrumentator import Instrumentator
Instrumentator().instrument(app).expose(app, endpoint="/metrics",
                                        include_in_schema=False)

⚠️ 三个必须记住的点：① uvicorn 会覆盖你的日志配置。uvicorn 启动时会执行自己的 LOGGING_CONFIG（在 uvicorn/config.py 里），它配置了 uvicorn、uvicorn.error、uvicorn.access 三个 logger，而且默认带 "disable_existing_loggers": True——这会把你在此之前创建的所有 logger 禁用掉。症状是「本地用 python main.py 跑日志正常，用 uvicorn main:app 跑就没了/格式不对」。两个解法：在自己的 dictConfig 里显式接管那三个 logger 并设 disable_existing_loggers: False（推荐，配置集中），或者用 uvicorn --log-config logging.json 传入自己的配置。② request_id 必须用 contextvars 而不是 threading.local。asyncio 下一个线程会交错执行多个协程，threading.local 存的值会在协程之间串——A 请求的日志带上 B 请求的 id，比没有 id 更糟糕。ContextVar 是为异步设计的：每个 Task 创建时会拷贝一份上下文快照，互不干扰。用完记得 reset(token)。③ 生产环境输出 JSON 到 stdout，不要写文件。容器里写文件有三个问题：容器重建就丢、多副本时日志散落在各处、磁盘写满会拖垮容器；而 JSON 格式让 ELK/Loki 能按字段查询和聚合（「找出所有 duration_ms > 1000 且 status = 500 的请求」），文本日志只能写正则。

完整版教学

一、uvicorn 的日志配置

★ ★uvicorn 的三个 logger★：
  ┌──────────────────┬────────────────────────────────┐
  │ ★uvicorn★         │ 通用                            │
  │ ★uvicorn.error★   │ ★启动信息、worker 生命周期、异常★│
  │ ★uvicorn.access★  │ ★每个请求的访问日志★             │
  └──────────────────┴────────────────────────────────┘
  ★ ★注意：uvicorn.error 记的不只是错误★（启动的 "Application startup
    complete" 也在这里）

★ ★★覆盖问题的根源★★：
  # uvicorn/config.py
  LOGGING_CONFIG = {
      "version": 1,
      ★"disable_existing_loggers": True★,     # ★★元凶★★
      "formatters": {...}, "handlers": {...},
      "loggers": {"uvicorn": {...}, "uvicorn.error": {...},
                  "uvicorn.access": {...}},
  }
  ★ uvicorn 启动时 → logging.config.dictConfig(LOGGING_CONFIG)
  → ★把此前创建的所有 logger 的 disabled 设为 True★
  → ★你的 logger 静默失效★

★ ★三种解法★：
  ① ★★在自己的 dictConfig 里接管（推荐）★★
     "loggers": {
         "uvicorn": {"handlers": ["console"], "level": "INFO",
                     ★"propagate": False★},
         "uvicorn.error": {...}, "uvicorn.access": {...},
     }
     ★ 并且自己的配置里 disable_existing_loggers: False
     ★ ★时机：在 uvicorn 加载 app 时执行★（放在 main.py 顶部或 create_app 里）

  ② ★--log-config 传文件★
     uvicorn main:app --log-config logging.yaml
     ★ ✓ 完全接管
     ★ ✗ 配置分散在代码外

  ③ ★启动后手动修复★（★不推荐，时机难控★）
     for name in ("uvicorn", "uvicorn.access", "uvicorn.error"):
         logging.getLogger(name).handlers = my_handlers

★ ★gunicorn + uvicorn worker 的情况★：
  gunicorn -k uvicorn.workers.UvicornWorker app:app
  ★ ★还多了 gunicorn.error / gunicorn.access 两个 logger★
  ★ ★而且 gunicorn 的 --log-config 和 uvicorn 的是两套★
  ✓ 统一在应用的 dictConfig 里接管所有五个

★ ★访问日志：用 uvicorn 的还是自己写★：
  ┌──────────────────────┬────────────────────────────┐
  │ ★uvicorn.access★      │ ✓ 开箱即用                  │
  │                       │ ✗ ★格式固定、信息少★         │
  │                       │ ✗ ★没有 request_id / user★  │
  │                       │ ✗ ★没有耗时（除非自己改）★   │
  │ ★自己写中间件★         │ ✓ ★结构化、字段自定义★       │
  │                       │ ✓ ★能带 request_id/user_id★ │
  │                       │ ✓ ★能过滤健康检查★           │
  └──────────────────────┴────────────────────────────┘
  ✓ ★生产建议：关掉 uvicorn.access（设 WARNING），自己写★

★ ★过滤健康检查（★很实用★）★：
  class HealthCheckFilter(logging.Filter):
      def filter(self, record):
          return "/health" not in record.getMessage()
  logging.getLogger("uvicorn.access").addFilter(HealthCheckFilter())
  ★ ★k8s 的探针每几秒一次，不过滤会淹没日志★

uvicorn 有三个 logger（uvicorn/uvicorn.error/uvicorn.access），注意 uvicorn.error 记的不只是错误（启动信息也在里面）。覆盖问题的根源是 uvicorn 的 LOGGING_CONFIG 里 disable_existing_loggers: True——它在启动时执行 dictConfig，把此前创建的所有 logger 禁用掉。推荐的解法是在自己的 dictConfig 里接管那三个 logger 并设 propagate: False。用 gunicorn + uvicorn worker 时还多了两个 gunicorn 的 logger，统一在应用配置里接管。访问日志建议自己写中间件——uvicorn 的格式固定、没有 request_id、没有 user_id、没有耗时；而且要过滤掉健康检查（k8s 探针每几秒一次，不过滤会淹没日志）。

二、request_id 与上下文传递

★ ★★为什么必须用 contextvars★★：
  ✗ threading.local：
    ★asyncio 下一个线程交错执行多个协程★
    → ★A 请求设置的值，B 请求可能读到★
    → ★★日志串号比没有 id 更糟★★
  ✓ ★ContextVar★：
    ★每个 asyncio.Task 创建时拷贝一份上下文快照★
    → 协程之间隔离
    → ★而且能穿透 await★（不像局部变量要层层传递）

★ ★完整实现★：
  from contextvars import ContextVar
  request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")
  user_id_ctx: ContextVar[str | None] = ContextVar("user_id", default=None)

  # ① 中间件里设置
  token = request_id_ctx.set(rid)
  try:
      await self.app(scope, receive, send_wrapper)
  finally:
      ★request_id_ctx.reset(token)★           # ★★用 token 精确恢复★★

  # ② 依赖里补充 user_id
  async def get_current_user(...):
      user = ...
      ★user_id_ctx.set(str(user.id))★
      return user

  # ③ Filter 注入
  class ContextFilter(logging.Filter):
      def filter(self, record):
          record.request_id = request_id_ctx.get()
          record.user_id = user_id_ctx.get() or "-"
          return True

★ ★★contextvars 的坑：run_in_threadpool★★：
  ★ anyio 的 to_thread.run_sync ★会传播 context★ ✓
  ★ 但 ★手动 threading.Thread 不会★ ✗
  ★ ★asyncio.create_task 会拷贝当前 context★ ✓
    → ★但任务里 set 的值不会影响外面★（是拷贝）
  ★ ★loop.run_in_executor 不会自动传播★ ✗
    ✓ 手动传：
      ctx = contextvars.copy_context()
      await loop.run_in_executor(pool, ctx.run, fn, arg)

★ ★跨服务传递（分布式追踪的雏形）★：
  # 调用下游时带上
  async def call_downstream():
      headers = {"X-Request-Id": request_id_ctx.get()}
      return await client.get(url, headers=headers)
  ★ ✓ ★整条调用链的日志能串起来★
  ★ ✓ 更标准的做法：★W3C Trace Context（traceparent 头）★
    → 用 OpenTelemetry 自动处理

★ ★传给后台任务★：
  # Celery
  @app.post("/x")
  async def x(bt: BackgroundTasks):
      rid = request_id_ctx.get()
      send_task.delay(data, ★request_id=rid★)     # ★★显式传★★
  # 任务里
  @celery.task
  def send_task(data, request_id=None):
      request_id_ctx.set(request_id or "-")
  ★ ★否则异步任务的日志和原始请求断开了★

★ ★回显给客户端（★排障关键★）★：
  response.headers["X-Request-Id"] = rid
  ★ ★用户报障时提供这个 ID → 你能精确定位那次请求的完整日志★
  ★ 500 响应里也要带：
    {"detail": "内部错误", "request_id": rid}

必须用 contextvars 而不是 threading.local——asyncio 下一个线程交错执行多个协程，threading.local 会让 A 请求的日志带上 B 请求的 id，比没有 id 更糟；ContextVar 每个 Task 创建时拷贝一份上下文快照，天然隔离，而且能穿透 await（不用层层传参）。有几个传播细节要知道：anyio.to_thread.run_sync 会传播 context（✓）、asyncio.create_task 会拷贝（任务里 set 的值不影响外面）、loop.run_in_executor 不会自动传播（要手动 copy_context()）。跨服务传递靠在请求头里带上 request_id（更标准的是 W3C Trace Context 的 traceparent 头，OpenTelemetry 会自动处理）；传给 Celery 任务要显式传参，否则异步任务的日志会和原始请求断开。回显给客户端是排障关键——用户报障时提供这个 ID 就能精确定位。

三、结构化日志与内容规范

★ ★结构化日志的价值★：
  ★文本★：[2026-08-03 10:00:00] INFO 用户 42 下单成功，金额 99.5，耗时 45ms
    → ★查"耗时>1000ms 且失败的请求"只能写正则★
  ★JSON★：
    {"ts":"...","level":"INFO","logger":"myapp.orders","msg":"order created",
     "request_id":"a1b2","user_id":42,"order_id":1001,"amount":99.5,
     "duration_ms":45,"path":"/api/orders","status":201}
    → ★ELK/Loki 里直接按字段查询和聚合★

★ ★实现（python-json-logger）★：
  "formatters": {"json": {
      "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
      "format": "%(asctime)s %(levelname)s %(name)s %(message)s "
                "%(request_id)s %(pathname)s %(lineno)d",
  }}
  # 记录额外字段
  logger.info("order created", ★extra={"order_id": 1001, "amount": 99.5}★)
  ★ ✗ 注意：★extra 的 key 不能和 LogRecord 内置属性冲突★
    （message / args / levelname / module / exc_info 等会报错）
  ★ 也可以用 structlog（更彻底的结构化日志方案）

★ ★★该记什么级别★★：
  ┌──────────┬────────────────────────────────────────┐
  │ DEBUG    │ ★开发排查★，生产关闭                     │
  │ ★INFO★    │ ★正常的业务事件★（登录、下单、任务完成） │
  │ ★WARNING★ │ ★可恢复的异常★（重试成功、降级、限流触发）│
  │ ★ERROR★   │ ★需要人关注的失败★                      │
  │ CRITICAL │ 服务不可用                              │
  └──────────┴────────────────────────────────────────┘
  ★ ★原则：ERROR 应该是"需要有人看一眼"的★
    → ★每天几万条 ERROR = 告警疲劳 = 等于没有告警★
  ★ ✗ ★参数校验失败（422）不该是 ERROR★（那是正常的用户输入错误）

★ ★★两条写法红线★★：
  ① ★用 logger.exception() 而不是 logger.error(str(e))★
     ✗ logger.error(f"失败: {e}")          # ★丢了堆栈★
     ✓ logger.exception("处理失败 order=%s", oid)   # ★★自动带 traceback★★
     ★ 必须在 except 块内调用
  ② ★用 %s 惰性格式化而不是 f-string★
     ✗ logger.debug(f"data={expensive_repr(obj)}")  # ★★总是先求值★★
     ✓ logger.debug("data=%s", obj)                 # ★级别不够时不格式化★
     ★ 附带好处：★Sentry 能按消息模板聚合同类错误★
       （f-string 会产生一万个不同的消息 = 一万个 issue）

★ ★★绝不能记录的内容★★：
  ✗ ★密码★（哪怕是错误的密码）
  ✗ ★token / API Key / SECRET_KEY★
  ✗ ★完整银行卡号 / 身份证号★
  ✗ ★Authorization / Cookie 头★
  ★ ★最常见的泄露点★：
    ① ★把整个请求体打进日志★
       logger.info("body=%s", await request.body())   # ★★含密码★★
    ② ★把请求头全打★（含 Authorization）
    ③ ★sqlalchemy.engine 的 INFO 会打印带参数的完整 SQL★
    ④ ★httpx 的 DEBUG 会打印完整请求头（含下游 API Key）★
    ⑤ ★422 的错误详情含 input 字段（回显用户输入）★
    ⑥ ★URL 里的敏感参数进 access log★
       → ★敏感参数放 POST body 而不是 query string★
  ✓ 脱敏：
    SENSITIVE = {"password", "token", "secret", "authorization", "card_no"}
    def mask(d): return {k: ("***" if k.lower() in SENSITIVE else v)
                         for k, v in d.items()}

结构化日志的价值是「可查询、可聚合」——JSON 格式能在 ELK/Loki 里按字段筛选（「找出耗时超过 1 秒且失败的请求」），文本日志只能写正则。记录级别的原则是「ERROR 应该是需要有人看一眼的」——每天几万条 ERROR 等于告警疲劳、等于没有告警；参数校验失败（422）不该是 ERROR（那是正常的用户输入错误）。两条写法红线：用 logger.exception() 而不是 logger.error(str(e))（前者自动带完整 traceback）、用 %s 惰性格式化而不是 f-string（级别不够时不求值，而且 Sentry 能按消息模板聚合同类错误）。六个泄露点都很隐蔽，最常见的是把整个请求体或请求头打进日志，还有 sqlalchemy.engine 的 INFO 会打印带参数的完整 SQL、httpx 的 DEBUG 会打印下游 API Key、以及 Pydantic 422 错误里的 input 字段会回显用户输入。

四、可观测性三支柱

★ ★三支柱★：
  ┌────────────┬────────────────────────────────────────┐
  │ ★Logs★      │ ★发生了什么★（离散事件，含上下文）        │
  │ ★Metrics★   │ ★有多少、多快★（聚合数值，适合告警和趋势）│
  │ ★Traces★    │ ★一次请求经过了哪些环节、各耗时多久★      │
  └────────────┴────────────────────────────────────────┘
  ★ ★三者靠 trace_id / request_id 关联★

★ ★★OpenTelemetry（最省事的方案）★★：
  pip install opentelemetry-distro opentelemetry-exporter-otlp
  opentelemetry-bootstrap -a install       # ★自动装所有可用的 instrumentation★

  from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
  from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
  from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
  from opentelemetry.instrumentation.redis import RedisInstrumentor

  FastAPIInstrumentor.instrument_app(app, ★excluded_urls="/health,/metrics"★)
  SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine)
  HTTPXClientInstrumentor().instrument()
  RedisInstrumentor().instrument()
  ★ → ★自动获得：★
    - ★每个路由的耗时和状态★
    - ★每条 SQL 的语句和耗时★
    - ★每个外部 HTTP 调用★
    - ★Redis 命令★
    - ★跨服务的 trace 传播（W3C traceparent 头）★

  # ★零代码方式★
  opentelemetry-instrument uvicorn main:app

★ ★手动加 span（业务关键路径）★：
  from opentelemetry import trace
  tracer = trace.get_tracer(__name__)

  async def process_order(order):
      with tracer.start_as_current_span("process_order") as span:
          ★span.set_attribute("order.id", order.id)★
          ★span.set_attribute("order.amount", float(order.amount))★
          try:
              await do_work()
          except Exception as e:
              ★span.record_exception(e)★
              ★span.set_status(Status(StatusCode.ERROR))★
              raise

★ ★把 trace_id 写进日志（★关联的关键★）★：
  class TraceFilter(logging.Filter):
      def filter(self, record):
          span = trace.get_current_span()
          ctx = span.get_span_context()
          record.trace_id = format(ctx.trace_id, "032x") if ctx.is_valid else "-"
          record.span_id = format(ctx.span_id, "016x") if ctx.is_valid else "-"
          return True
  ★ → ★在 APM 里看到一个慢请求，能直接跳到对应的日志★

★ ★Prometheus 指标★：
  from prometheus_fastapi_instrumentator import Instrumentator
  Instrumentator(
      excluded_handlers=["/health", "/metrics"],
  ).instrument(app).expose(app, endpoint="/metrics", include_in_schema=False)
  ★ → ★自动暴露：http_requests_total、http_request_duration_seconds
    （含 method/path/status 标签）★

  # ★自定义业务指标★
  from prometheus_client import Counter, Histogram, Gauge
  orders_created = Counter("orders_created_total", "订单数", ["channel"])
  order_amount = Histogram("order_amount", "订单金额", buckets=[10,50,100,500])
  active_streams = Gauge("active_streams", "进行中的流式请求")
  orders_created.labels(channel="web").inc()

★ ★★指标的基数陷阱★★：
  ✗ Counter("requests", labels=["user_id"])     # ★★100 万用户 = 100 万时间序列★★
  ✗ 把 path 里的 id 当标签：/orders/12345
    → ★Prometheus 内存爆炸★
  ✓ ★用路由模板而不是实际路径★：/orders/{id}
  ✓ ★标签的取值集合必须是有限且较小的★（method、status、endpoint）

★ ★健康检查与就绪探针★：
  @app.get("/health", include_in_schema=False)      # ★★liveness：只看进程活着★★
  async def health(): return {"status": "ok"}

  @app.get("/ready", include_in_schema=False)       # ★★readiness：依赖也要就绪★★
  async def ready(db: DbDep, redis: RedisDep):
      await db.execute(text("SELECT 1"))
      await redis.ping()
      return {"status": "ready"}
  ★ ✗ ★liveness 里别查数据库★——数据库抖动会导致 k8s ★重启所有 pod★

可观测性三支柱是「日志（发生了什么）、指标（有多少多快）、追踪（经过了哪些环节）」，三者靠 trace_id 关联。OpenTelemetry 的自动埋点是最省事的方案——几行代码就能拿到路由耗时、SQL 语句、外部 HTTP 调用、Redis 命令，还自动处理跨服务的 trace 传播；关键是把 trace_id 写进日志，这样在 APM 里看到慢请求能直接跳到对应日志。指标有个基数陷阱：标签里放 user_id 或实际路径（含 id）会让时间序列爆炸，必须用路由模板 /orders/{id} 而不是 /orders/12345。健康检查要区分 liveness 和 readiness——liveness 里千万别查数据库，否则数据库抖动会导致 k8s 重启所有 pod（雪上加霜）。

五、生产实践

★ ★容器化的日志★：
  ★ 输出到 ★stdout/stderr★，不写文件
  ★ 理由：
    ① ★容器重建就丢★
    ② ★多副本时散落在各处★
    ③ ★磁盘写满拖垮容器★
  ★ 收集：Docker logging driver / Fluent Bit / Promtail → Loki/ES

★ ★多 worker 的日志★：
  ★ 每个 worker 是独立进程，日志会交错
  ✓ ★靠 request_id 区分★
  ✓ 或在日志里带 ★process_id★：
    "format": "... [pid:%(process)d] ..."

★ ★日志量控制（★成本★）★：
  ① ★生产 INFO 级别★，DEBUG 只在排查时临时开
  ② ★sqlalchemy.engine 设 WARNING★（INFO 会打印每条 SQL）
  ③ ★uvicorn.access 关掉，自己写★（还能过滤健康检查）
  ④ ★采样★：高频接口的成功日志只记 1%
     if random.random() < 0.01 or status >= 400:
         logger.info(...)
  ⑤ ★大对象截断★：logger.info("data=%s", str(obj)[:500])
  ★ ★日志费用在云上是真金白银★（按 GB 计费）

★ ★告警设计★：
  ┌────────────────────┬──────────────────────────────┐
  │ ★P0（立即处理）★    │ 服务不可用、错误率 > 5%        │
  │ ★P1（工作时间）★    │ ★P99 恶化、特定接口失败率高★   │
  │ ★P2（每日汇总）★    │ 慢查询、告警趋势               │
  └────────────────────┴──────────────────────────────┘
  ★ ★别把所有 ERROR 都告警★ → 告警疲劳
  ✓ ★按"错误率"而不是"错误数"告警★（流量波动时更稳）
  ✓ ★关键路径单独监控★（登录、支付、下单）

★ ★Sentry 集成★：
  import sentry_sdk
  from sentry_sdk.integrations.fastapi import FastApiIntegration
  from sentry_sdk.integrations.logging import LoggingIntegration
  sentry_sdk.init(
      dsn=settings.SENTRY_DSN,
      environment=settings.ENV,
      ★traces_sample_rate=0.1★,              # ★性能追踪采样★
      integrations=[FastApiIntegration(),
                    LoggingIntegration(level=logging.INFO,      # ★面包屑★
                                       event_level=logging.ERROR)],  # ★上报★
      ★before_send=scrub_sensitive★,          # ★★脱敏★★
  )
  def scrub_sensitive(event, hint):
      if "request" in event:
          event["request"].pop("cookies", None)
          headers = event["request"].get("headers", {})
          headers.pop("Authorization", None)
      return event

★ ★排查线上问题的标准流程★：
  ① ★用户提供 request_id★（响应头里回显的）
  ② ★在日志系统里查这个 id → 拿到完整链路★
  ③ ★有 trace_id 的话跳到 APM 看各环节耗时★
  ④ ★Sentry 里看有没有对应的异常和堆栈★
  ⑤ ★指标看是个例还是普遍问题★
  ★ ★这套组合拳的前提就是：日志里有 request_id★

★ ★优雅关闭时的日志★：
  @asynccontextmanager
  async def lifespan(app):
      logger.info("应用启动 version=%s", VERSION)
      yield
      logger.info("应用关闭中...")            # ★★确认收到了 SIGTERM★★
      await cleanup()
      logger.info("应用已关闭")
  ★ ★部署时能确认 pod 是优雅退出还是被强杀★

生产实践的核心是容器里输出到 stdout、多 worker 靠 request_id 区分、控制日志量（生产 INFO、sqlalchemy.engine 设 WARNING、关掉 uvicorn.access 自己写、高频接口采样——日志费用在云上是真金白银）。告警要分级，而且按「错误率」而不是「错误数」（流量波动时更稳），别把所有 ERROR 都告警。Sentry 集成时记得配 before_send 做脱敏（去掉 cookies 和 Authorization 头）。排查线上问题的标准流程是「用户给 request_id → 查日志拿完整链路 → 跳 APM 看耗时 → Sentry 看堆栈 → 指标看是个例还是普遍」——而这套组合拳的前提就是日志里有 request_id。

六、实践清单

★ 完整的日志模块（可直接抄）：
  # myapp/logging_ext.py
  from contextvars import ContextVar
  import logging, uuid, time

  request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")

  class ContextFilter(logging.Filter):
      def filter(self, record):
          record.request_id = request_id_ctx.get()
          return True

  def setup_logging(is_prod: bool):
      logging.config.dictConfig({
          "version": 1, ★"disable_existing_loggers": False★,
          "filters": {"ctx": {"()": ContextFilter}},
          "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"},
          },
          "handlers": {"console": {"class": "logging.StreamHandler",
                                   "stream": "ext://sys.stdout",
                                   "formatter": "json" if is_prod else "text",
                                   "filters": ["ctx"]}},
          "root": {"level": "INFO", "handlers": ["console"]},
          "loggers": {
              ★"uvicorn": {"handlers": ["console"], "propagate": False}★,
              ★"uvicorn.error": {"handlers": ["console"], "propagate": False}★,
              ★"uvicorn.access": {"level": "WARNING", "propagate": False}★,
              "sqlalchemy.engine": {"level": "WARNING"},
          },
      })

★ 检查清单：
  【配置】
  □ ★setup_logging() 在 create_app 之前调用★
  □ ★disable_existing_loggers: False★
  □ ★接管了 uvicorn 的三个 logger★
  □ ★sqlalchemy.engine 设 WARNING★
  □ ★生产 JSON 输出到 stdout★
  【request_id】
  □ ★用 contextvars 不是 threading.local★
  □ ★中间件里 set 并在 finally 里 reset★
  □ ★Filter 注入且 return True★
  □ ★响应头回显 X-Request-Id★
  □ ★调下游服务时透传★
  □ ★传给 Celery 任务★
  【内容】
  □ ★logger.exception 而不是 error(str(e))★
  □ ★%s 惰性格式化★
  □ ★不记密码/token/完整卡号★
  □ ★不整个打印请求体和请求头★
  □ ★422 的 input 字段过滤掉★
  【可观测】
  □ ★OTel 自动埋点（排除 /health）★
  □ ★trace_id 写进日志★
  □ ★Prometheus 指标（标签基数可控）★
  □ ★liveness 不查数据库★
  □ ★告警按错误率分级★

★ ★排查"看不到日志"的顺序★：
  ① ★是不是被 uvicorn 的 disable_existing_loggers 禁用了★
  ② ★logger 的有效级别★：getEffectiveLevel()
  ③ ★handler 的级别★
  ④ ★propagate 是否被关★
  ⑤ ★配置时机（在 uvicorn 启动前？）★
  ⑥ ★输出目标是 stdout 吗（容器里才能被采集）★

★ 一句话总结：
  ★"FastAPI 用的就是标准库 logging，但 uvicorn 启动时会用
    disable_existing_loggers=True 覆盖你的配置——要在自己的
    dictConfig 里接管 uvicorn 的三个 logger；
    request_id 必须用 contextvars（threading.local 在协程间会串），
    中间件里 set、Filter 注入、响应头回显、调下游时透传；
    生产输出 JSON 到 stdout，可观测性用 OTel 自动埋点。"★

检查清单分四块。排查「看不到日志」的第一步就是「是不是被 uvicorn 的 disable_existing_loggers 禁用了」——这是 FastAPI 项目特有的头号原因，其次才是常规的级别、handler、propagate、配置时机。

记忆钩子：「FastAPI 本身没有日志系统，★用的就是标准库 logging★——但它跑在 uvicorn 里，而 ★uvicorn 启动时会执行自己的 LOGGING_CONFIG，其中带 disable_existing_loggers: True★，★把你此前创建的所有 logger 禁用掉★——★这就是『明明配了 dictConfig 日志却不听话』『用 python main.py 跑正常、用 uvicorn 跑就没了』的根本原因★。解法是★在自己的 dictConfig 里显式接管 uvicorn / uvicorn.error / uvicorn.access 三个 logger（设 propagate: False）并写 disable_existing_loggers: False★，或用 —log-config 传文件；★gunicorn + uvicorn worker 时还多两个 gunicorn logger★。注意 ★uvicorn.error 记的不只是错误★（启动信息也在里面），★生产建议关掉 uvicorn.access 自己写中间件★（它格式固定、没有 request_id/user_id/耗时，而且★要过滤健康检查否则 k8s 探针会淹没日志★）。★第二个核心是 request_id，必须用 contextvars 而不是 threading.local★——★asyncio 下一个线程交错执行多个协程，threading.local 会让 A 请求的日志带上 B 的 id，比没有 id 更糟★；ContextVar ★每个 Task 创建时拷贝上下文快照★所以天然隔离，还★能穿透 await★。实现四步：★中间件里 set（并在 finally 里 reset(token)）→ logging.Filter 注入（filter() 必须 return True）→ 响应头回显 X-Request-Id → 调下游和投 Celery 时显式透传★。传播细节：★anyio.to_thread.run_sync 会传播 context、asyncio.create_task 会拷贝、但 loop.run_in_executor 不会（要手动 copy_context()）★。★生产输出 JSON 到 stdout★（容器里写文件会丢、会散落、会写满磁盘），JSON 让 ELK/Loki ★能按字段查询聚合★。两条写法红线：★logger.exception() 而不是 error(str(e))★（前者带完整堆栈）、★%s 惰性格式化而不是 f-string★（级别不够时不求值，★而且 Sentry 能按消息模板聚合同类错误★）。★六个泄露点★：整个请求体、整个请求头（含 Authorization）、★sqlalchemy.engine 的 INFO 会打印带参数的完整 SQL★、★httpx 的 DEBUG 会打印下游 API Key★、★Pydantic 422 错误的 input 字段回显用户输入★、★URL 里的敏感参数进 access log（所以敏感参数放 POST body）★。可观测性三支柱★日志/指标/追踪靠 trace_id 关联★，★OpenTelemetry 自动埋点最省事★（几行拿到路由耗时+SQL+外部 HTTP+Redis，还自动传播 W3C traceparent），★关键是把 trace_id 写进日志才能从 APM 跳到日志★。两个陷阱：★指标标签基数爆炸（别放 user_id 或含 id 的实际路径，要用路由模板 /orders/{id}）★、★liveness 探针里别查数据库（数据库抖动会让 k8s 重启所有 pod）★。」

七、常见误区与追问

误区：配好 logging.dictConfig 日志就会按我的格式输出。 uvicorn 会覆盖它。uvicorn 在启动时会执行自己的 LOGGING_CONFIG（定义在 uvicorn/config.py 里），而这个配置带着 "disable_existing_loggers": True——dictConfig 的这个选项会把当前已存在的、且不在新配置里声明的所有 logger 的 disabled 属性设为 True，于是你精心配置的 myapp logger 就静默失效了。典型症状是「用 python main.py 直接跑日志正常，用 uvicorn main:app 跑就没了或者变回默认格式」，非常迷惑。两个解法：① 在自己的 dictConfig 里显式声明 uvicorn、uvicorn.error、uvicorn.access 三个 logger（指定 handler 和 propagate: False），同时自己的配置里写 disable_existing_loggers: False——这样即使 uvicorn 后执行也是覆盖到你已经接管过的配置上；② 用 uvicorn --log-config logging.yaml 把配置完全交给 uvicorn 加载。用 gunicorn + UvicornWorker 时还要一并接管 gunicorn.error 和 gunicorn.access。
误区：用 threading.local 存 request_id，反正每个请求一个线程。 在 asyncio 下这个前提不成立。FastAPI 的 async def 路由全部跑在同一个事件循环线程里——一个线程会交错执行几十上百个协程。threading.local 的隔离粒度是「线程」，所以所有并发请求共享同一份数据：A 请求刚 set 了自己的 id，切到 B 请求又 set 了新的，等切回 A 时读到的已经是 B 的 id 了。结果是日志里的 request_id 完全错乱——这比根本没有 id 更糟糕，因为它会把你的排查引向完全错误的请求。正确的是 contextvars.ContextVar：它是专为异步设计的，每个 asyncio.Task 创建时会拷贝一份当前上下文的快照，之后各自修改互不影响。用完记得 reset(token)（虽然 Task 结束时上下文会自动销毁，但在中间件这种复用同一个上下文的场景下显式 reset 更稳妥）。
误区：logger.error(f"处理失败: {e}") 已经把异常信息记下来了。 丢掉了最关键的堆栈。str(e) 只有异常的消息文本，很多时候还很含糊（KeyError: 'user_id'、IntegrityError），你完全不知道它是在哪一行、经过什么调用路径抛出来的——排查时只能靠猜或者去翻代码。正确写法是 logger.exception("处理失败 order_id=%s", oid)——它等价于 logger.error(..., exc_info=True)，会自动把完整的 traceback 附在日志记录里（必须在 except 块内调用）。在异步代码里这一点尤其重要，因为协程的调用栈本来就比同步代码难还原。如果不在 except 块内但手上有异常对象，可以用 logger.error("...", exc_info=e)。另外提醒：FastAPI 对未被 exception_handler 处理的异常会自己记录，但如果你注册了全局的 @app.exception_handler(Exception)，就必须在里面自己调用 logger.exception()，否则堆栈就彻底丢了。
误区：日志里打印请求体方便排查问题。 这是最常见的敏感信息泄露点。logger.info("body=%s", await request.body()) 这一行，在登录接口上就直接把明文密码记进了日志——而日志通常保留数月、被多人访问、还可能同步到第三方平台（Sentry、ELK、云日志服务）。同类的隐蔽泄露还有五个：① 打印全部请求头（含 Authorization 和 Cookie）；② sqlalchemy.engine 设成 INFO 会打印带绑定参数的完整 SQL，注册接口的那条 INSERT 里就有密码；③ httpx/urllib3 的 DEBUG 日志会打印完整请求头，把你调用下游服务的 API Key 泄露出去；④ Pydantic V2 的 422 错误里有 input 字段，会回显用户提交的原始值——如果用户在 password 字段传错了类型，密码就出现在错误响应和日志里；⑤ URL 里的敏感参数会进 access log（所以敏感参数要放 POST body 而不是 query string）。防御是双层的：主要靠「一开始就不记」（只记 id 不记内容、用脱敏字典过滤），辅以全局 logging.Filter 做正则兜底。
误区：给 Prometheus 指标加上 user_id 标签，方便按用户分析。 会导致时间序列爆炸（高基数问题）。Prometheus 的每一个「指标名 + 标签值组合」都是一条独立的时间序列，要在内存里维护——http_requests_total{user_id="12345"} 这样的指标，有多少用户就有多少条序列，100 万用户就是 100 万条，Prometheus 的内存会直接爆掉，查询也会变得极慢。同样的问题出现在把实际请求路径当标签：/orders/12345、/orders/12346……每个订单 id 都产生一条新序列。正确做法是用路由模板：FastAPI 的 request.scope["route"].path 能拿到 /orders/{id} 这种模板形式（prometheus-fastapi-instrumentator 默认就是这么做的）。标签的取值集合必须是有限且较小的——method（约 7 种）、status（几十种）、endpoint（几十到几百）都没问题。需要按用户分析的话，那属于日志或数据仓库的职责，不是指标的。
追问：为什么建议关掉 uvicorn.access 自己写访问日志？ 因为 uvicorn 自带的访问日志信息太少且不可扩展。它的格式是固定的（客户端IP - "METHOD /path HTTP/1.1" 状态码），没有请求耗时（这是最关键的性能指标）、没有 request_id（无法和应用日志关联）、没有 user_id（不知道是谁的请求）、不是结构化的（日志系统里难以按字段查询）、也无法过滤（k8s 的存活探针每几秒请求一次 /health，这些噪音会淹没真正有价值的日志，还平白增加日志存储成本）。自己写一个纯 ASGI 中间件只要二十行，就能记录：method、path（用路由模板）、status、duration_ms、client_ip（配好 ProxyHeaders 后是真实 IP）、request_id、user_id，并且跳过健康检查路径、对高频成功请求做采样。做法是把 uvicorn.access 的级别设成 WARNING（或 propagate: False 且不给 handler）关掉它，然后注册自己的中间件。
追问：OpenTelemetry 值得引入吗？相比自己埋点有什么优势？ 在有多个服务的场景下非常值得，单体应用也有价值。核心优势是自动埋点：FastAPIInstrumentor.instrument_app(app) 加上 SQLAlchemy、httpx、Redis 的 instrumentation，几行代码就能拿到「一次请求的完整分解」——路由处理了多久、其中哪几条 SQL 各花了多久、调用下游服务耗时多少、Redis 命令多少次——这些信息自己埋点要写几百行且容易漏。第二个优势是跨服务传播：它遵循 W3C Trace Context 标准，自动在出站请求里注入 traceparent 头、在入站请求里解析，所以微服务之间的调用链天然串起来。第三个是厂商中立：同一套 SDK 可以导出到 Jaeger、Tempo、Datadog、SkyWalking、阿里云 ARMS，换后端不用改代码。落地的关键动作有三个：① 排除健康检查和指标端点（excluded_urls="/health,/metrics"，否则 trace 数据里全是噪音）；② 配置采样率（生产环境全量采集成本很高，通常 1%~10%，但错误请求要全量采样）；③ 把 trace_id 写进日志——这样在 APM 里发现一个慢请求后，能直接用 trace_id 跳到对应的日志详情，这是三支柱关联的关键一步。
追问：健康检查接口该怎么设计？ 要区分 liveness（存活）和 readiness（就绪），而且两者的检查内容完全不同。/health（liveness）只应该检查「进程还活着、事件循环没卡死」——返回一个简单的 {"status": "ok"} 就够了。绝对不要在 liveness 探针里检查数据库或 Redis：因为 liveness 失败时 k8s 会重启这个 pod，而数据库短暂抖动会让所有 pod 的 liveness 同时失败 → 全部重启 → 连接风暴 → 数据库彻底崩溃，是典型的雪上加霜。/ready（readiness）则要检查依赖是否可用（数据库 SELECT 1、Redis ping、必要的下游服务），因为 readiness 失败时 k8s 只是把这个 pod 从负载均衡里摘掉、不再给它转发流量，等它恢复了再加回来——这正是我们想要的行为。三个补充细节：① 两个端点都要 include_in_schema=False（不出现在 API 文档里）；② 都要从访问日志里过滤掉（探针频率很高）；③ readiness 检查要设短超时（比如 2 秒），否则探针自己会超时挂起。

八、加强记忆

FastAPI 本身没有日志系统，用的就是标准库 logging——但它跑在 uvicorn 里，而 uvicorn 启动时会执行自己的 LOGGING_CONFIG，其中带着 disable_existing_loggers: True，把你此前创建的所有 logger 禁用掉——这就是「明明配了 dictConfig 日志却不听话」「用 python main.py 跑正常、用 uvicorn 跑就没了」的根本原因。解法是在自己的 dictConfig 里显式接管 uvicorn/uvicorn.error/uvicorn.access 三个 logger（设 propagate: False）并写 disable_existing_loggers: False，或者用 --log-config 传文件；用 gunicorn + uvicorn worker 时还多两个 gunicorn 的 logger。注意 uvicorn.error 记的不只是错误（启动信息也在里面），生产建议关掉 uvicorn.access 自己写中间件（它格式固定、没有 request_id/user_id/耗时，而且要过滤健康检查，否则 k8s 探针会淹没日志）。第二个核心是 request_id，必须用 contextvars 而不是 threading.local——asyncio 下一个线程交错执行多个协程，threading.local 会让 A 请求的日志带上 B 请求的 id，比没有 id 更糟糕；ContextVar 每个 Task 创建时拷贝一份上下文快照所以天然隔离，还能穿透 await。实现分四步：中间件里 set（并在 finally 里 reset(token)）→ logging.Filter 注入（filter() 必须 return True）→ 响应头回显 X-Request-Id → 调下游和投递 Celery 时显式透传。传播细节：anyio.to_thread.run_sync 会传播 context、asyncio.create_task 会拷贝、但 loop.run_in_executor 不会（要手动 copy_context()）。生产环境输出 JSON 到 stdout（容器里写文件会丢、会散落、会写满磁盘），JSON 格式让 ELK/Loki 能按字段查询和聚合。两条写法红线：logger.exception() 而不是 error(str(e))（前者带完整堆栈）、%s 惰性格式化而不是 f-string（级别不够时不求值，而且 Sentry 能按消息模板聚合同类错误）。六个泄露点：整个请求体、整个请求头（含 Authorization）、sqlalchemy.engine 的 INFO 会打印带参数的完整 SQL、httpx 的 DEBUG 会打印下游 API Key、Pydantic 422 错误的 input 字段会回显用户输入、URL 里的敏感参数会进 access log（所以敏感参数要放 POST body）。可观测性三支柱日志/指标/追踪靠 trace_id 关联，OpenTelemetry 的自动埋点最省事（几行就能拿到路由耗时 + SQL + 外部 HTTP + Redis，还自动传播 W3C traceparent），关键是把 trace_id 写进日志才能从 APM 跳到日志。最后两个陷阱：指标标签的基数会爆炸（别放 user_id 或含 id 的实际路径，要用路由模板 /orders/{id}）、liveness 探针里别查数据库（数据库抖动会让 k8s 重启所有 pod）。

简化版

详细版