用 LogBrew 处理 Flask 日志和 Flask 异常

使用 logbrew-flask 0.1.6 或更高版本。它会采集请求 Span（每个 Span 是一次计时操作）和未处理的异常。可选的日志处理器会把 Python 日志关联到请求链路。请求耗时指标同样是可选的。该包需要 Python 3.10 或更高版本，以及 Flask 3.1 或更高版本。

在应用中使用项目级的服务端接入密钥。不要使用为 CLI 授权的账号凭据。

把 <project_id>、<trace_id> 和 <issue_id> 替换为 LogBrew 中的 ID。测试请求使用开发项目。Shell 示例使用 Bash。

发送第一个请求

创建或选择项目密钥

使用 CLI 登录。然后使用新项目或已有项目，两个示例只运行其中一个。对于新项目，把它的接入密钥保存到一个仅所有者可访问的新文件中：

install -d -m 700 "$HOME/.logbrew"
logbrew projects create my-flask-app \
  --runtime flask \
  --environment development \
  --ingest-key-file "$HOME/.logbrew/my-flask-app.ingest" \
  --json

对于已有项目，从 logbrew projects --json 获取项目 ID，然后创建一个单独的服务端密钥：

logbrew projects keys create "<project_id>" \
  --kind server \
  --label "Flask server" \
  --ingest-key-file "$HOME/.logbrew/my-flask-app.ingest" \
  --json

CLI 以仅所有者可访问的权限保存接入密钥，不会输出密钥或文件路径。在生产环境中，使用部署平台的密钥管理服务，而不是这个本地文件。如果创建请求的结果不确定，原样再运行一次该命令。

安装已发布的集成

创建或激活应用的虚拟环境，然后安装框架包。它会同时安装兼容的核心 SDK 依赖。

python3 -m pip install "logbrew-flask>=0.1.6,<0.2"

配置部署标识

在本地加载项目密钥，不要输出它。设置稳定的服务、环境和版本，让每个信号都带有有用的部署上下文。

export LOGBREW_SERVER_API_KEY="$(tr -d '\r\n' < "$HOME/.logbrew/my-flask-app.ingest")"
export LOGBREW_SERVICE_NAME="my-flask-app"
export LOGBREW_ENVIRONMENT="development"
export LOGBREW_RELEASE="my-flask-app@1.0.0"

init_logbrew() 先检查显式传入的参数，然后是 Flask 应用配置，最后是这些环境变量。如果没有配置服务名，它会使用 Flask 的导入名称。

初始化 Flask（仅一次）

对于新应用，把示例保存为 app.py。创建 Flask 应用之后，再添加初始化代码。日志处理器是可选的，它让应用日志共享当前活动请求的链路和 Span。

import logging

from flask import Flask
from logbrew_flask import init_logbrew
from logbrew_sdk import LogBrewLoggingHandler

app = Flask(__name__)
logbrew = init_logbrew(app, capture_request_metrics=True)

logger = logging.getLogger("my-flask-app")
logger.setLevel(logging.INFO)
logger.addHandler(LogBrewLoggingHandler(logbrew.client))


@app.get("/health")
def health() -> dict[str, bool]:
    logger.info("health check completed")
    return {"ok": True}

重复调用 init_logbrew(app) 会返回同一个 Flask 扩展，不会重复安装请求钩子或异常钩子。初始化代码不会添加日志处理器。把处理器添加到应用自己的 logger 上，且只添加一次。

发送一个请求

启动应用，然后在另一个终端中调用该路由。如果你的应用使用其他模块名，替换命令中的 app。

flask --app app run

curl -fsS http://127.0.0.1:5000/health

投递在后台 worker 中进行，不会阻塞响应。

验证 LogBrew 中的证据

按确切的项目和部署范围读取。查找 health 请求的链路，以及示例日志处理器发出的关联日志。

logbrew read traces \
  --project "<project_id>" \
  --service my-flask-app \
  --environment development \
  --release my-flask-app@1.0.0 \
  --since 1h \
  --json

logbrew read logs \
  --project "<project_id>" \
  --service my-flask-app \
  --environment development \
  --release my-flask-app@1.0.0 \
  --search "health check completed" \
  --since 1h \
  --json

把返回的 Trace ID 复制到这条限定范围的排查命令中：

logbrew explain trace "<trace_id>" --json

读取到匹配的结果，才能确认 LogBrew 已存储该信号。本地返回 200 只能证明 Flask 处理了该请求。

选择你需要的信号

初始化后，请求链路和未处理异常的采集默认启用。请求指标和 Python 日志转发需要单独选择。

选项	默认值	适用场景
`capture_successful_requests`	`True`	你希望每个已完成的请求都有一个 Span。关闭此选项后，状态码为 `500` 或更高的响应仍会创建 Span。
`capture_exceptions`	`True`	你希望 Flask 报告为未处理的异常生成一个带类型的问题，以及对应的错误 Span。
`capture_request_metrics`	`False`	你希望为已完成的请求生成耗时直方图。
`LogBrewLoggingHandler`	未挂载	你希望把选定的 Python 日志条目转为 LogBrew 日志。
`flush_on_response`	`False`	使用后台投递时保持此默认值。只有在受控诊断中、需要暴露响应路径上的投递时才启用。

先使用默认值。需要查看延迟时，再添加请求指标。只给有助于排查的 logger 添加日志处理器。默认情况下，初始化代码会创建并管理一个 HTTP 传输层。它把每个信号放入队列，只要队列中有信号，后台 worker 就会开始投递。投递不会阻塞响应。传输失败不会影响 Flask 响应。 flush_on_response=True 会改为在响应路径上发送队列中的信号。启用该选项后，raise_flush_errors=True 会抛出投递错误。这两个选项只在受控诊断中使用。

控制日志内容

给产生你所需消息的 logger 添加一个 LogBrewLoggingHandler。该处理器不会替换应用的终端输出处理器或文件处理器，也不会修改 Flask 或 Python logging 模块。同一个处理器添加两次会导致日志重复。如果应用还使用了 root logger，挂载第二个处理器之前，先检查日志的传播设置。请求内的日志会继承当前活动的链路和 Span。请求外的日志没有请求链路。处理器会发送格式化后的消息、logger 名称和 LogBrew 级别，还会发送源文件名、模块、函数、行号、线程名和进程名，但不发送完整的源文件路径。extra={...} 中的基本类型值会成为元数据。在面向用户的请求中记录日志之前，先检查这些值。 Python 的 DEBUG 日志条目会转为 LogBrew 的 info。WARNING 转为 warning，CRITICAL 转为 critical。原始的 Python 级别名称和数值保留在元数据中。要减少发送的条目，为处理器单独设置级别，例如 LogBrewLoggingHandler(logbrew.client, level=logging.WARNING)。终端和 LogBrew 可以收到同一条日志条目。例如，在创建 Flask 应用之前配置终端输出：

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

终端收到的是格式化后的文本。LogBrew 则把消息、级别、logger 和元数据保存为独立的字段。带有 exc_info 的日志条目包含异常类型和消息。只有当处理器设置了 include_exception_text=True 时，才会发送完整的异常文本。这与自动异常采集是两回事，自动异常采集从不发送异常消息。不要在日志消息或异常消息中放入凭据、个人信息或请求体。

集成会采集什么

信号	默认行为
请求链路	一个服务端 Span，包含方法、Flask 路由模板、状态码和耗时。
异常问题	标题为 `<METHOD> <route template> failed`，包含未处理异常的类型、固定摘要 `Unhandled exception`、`flask.middleware` 机制、是否已处理的状态，以及最多 32 个经过清理的堆栈帧。
应用日志	通过 `LogBrewLoggingHandler` 启用；请求内的日志会继承当前活动的链路和 Span。
请求指标	通过 `capture_request_metrics=True` 启用；发出 `http.server.duration` delta 直方图，描述为 `Duration of one completed server request.`
入站链路	用一个新的子 Span 延续有效的 W3C `traceparent`。

Span 名称由 HTTP 方法和匹配到的路由模板组成。例如，/orders/100 和 /orders/200 共用 GET /orders/<int:order_id>。未匹配的请求使用固定标签 <unmatched>。traceparent 缺失或无效时，会开启一条新链路。 Span 元数据包含方法、路由模板、端点、状态码和耗时，不包含具体路径或查询字符串。在 0.1.6 及更高版本中，请求 Span 覆盖完整的 Flask 分发过程。before-request 钩子、错误处理器和 teardown 都共享请求链路。流式响应会在响应体发送完毕或关闭时结束其 Span。

测试异常采集

4xx 或 5xx 状态会显示在请求 Span 上。仅凭状态码不能证明发生了异常。诊断故障之前，链路和问题都要查看。LogBrew 不会替换 Flask 错误处理器，也不会更改响应状态。在开发项目中添加一个临时路由，调用一次。部署之前删除它：

@app.get("/logbrew-test-error")
def handle_test_error() -> None:
    raise RuntimeError("local capture check")

该请求返回状态码 500。查找新的问题和请求指标：

logbrew read issues --project "<project_id>" --since 1h --json
logbrew explain issue "<issue_id>" --json

logbrew explain metric http.server.duration \
  --project "<project_id>" \
  --service my-flask-app \
  --environment development \
  --release my-flask-app@1.0.0 \
  --since 1h \
  --json

检查该问题的服务、环境、版本、异常类型和请求链路。问题中不能包含文本 local capture check。

把出站工作关联到请求

Python SDK 为 requests、urllib、httpx、aiohttp、数据库、缓存和队列工作提供了由应用调用的辅助函数。每个辅助函数都会在当前活动的请求链路中创建一个子 Span。SDK 不会全局修改这些库。logbrew-flask 包提供了示例。为每个辅助函数提供稳定的操作名，以及路由模板或语句模板。不要在这些名称中放入记录 ID、客户数据值、SQL 绑定值或消息负载。只在依赖对请求有影响的地方添加子 Span。为每个小调用都创建 Span，会让链路更难阅读。出站 HTTP 辅助函数会添加一个 W3C traceparent 请求头，不会采集完整 URL、查询字符串、请求体、Cookie 或任意请求头。数据库、缓存和队列辅助函数不会采集绑定值、查询结果、缓存值或队列负载。

隐私与进程生命周期

Flask 自动采集不包含具体的请求路径、查询字符串、请求体和响应体、Cookie、任意请求头、原始 traceparent 值、baggage 和 tracestate。自动异常采集从不读取或发送 str(error)。它使用固定摘要 Unhandled exception，并把异常链中的每条消息标记为已隐藏。堆栈帧包含文件基本名称和有限的代码标识，不包含原始 traceback 文本、源代码、局部变量或绝对路径。在每个 worker 进程内创建应用和 LogBrew 客户端。Gunicorn 默认按 worker 加载应用，这种方式可以正常工作。如果客户端是在父进程中创建的，不要使用 Gunicorn 的 --preload。对于 uWSGI，使用按 worker 加载应用的方式，例如 --lazy-apps。进程 fork 之后，Python 客户端会拒绝被复用。在进程管理器的 worker 关闭钩子中调用 logbrew.client.shutdown()。它会发送队列中剩余的信号并关闭客户端。不要在每个请求之后调用它。关闭之后，客户端会拒绝该 worker 的新信号。响应可能在 worker 发送最后一批排队信号之前就已结束。在部署或缩容期间，优雅关闭会让这些信号获得最后一次投递机会。

排除接入故障

症状	检查	下一步
`LOGBREW_SERVER_API_KEY is required`	初始化代码没有收到非空的项目密钥。	通过 Flask 配置、显式参数或环境变量配置项目级服务端密钥。
没有出现链路	项目、服务、环境、版本或时间筛选条件不匹配。	运行 `logbrew doctor --project "<project_id>" --json`，然后按范围重新读取链路。
有链路但没有日志	请求插桩不会自动挂载 Python 日志处理器。	给应用自己的 logger 添加一个 `LogBrewLoggingHandler`。
请求正常，但关闭时最后的信号丢失	进程在后台 worker 发送完剩余数据之前就退出了。	在 worker 关闭钩子中调用 `logbrew.client.shutdown()`。
worker 报告 `process_ownership_error`	客户端在 worker 进程 fork 之前就已创建。	按 worker 加载应用，并在该进程中创建新的客户端。

账号访问、密钥和回读错误，参见故障排除。其他运行时，参见 SDK 接入。

开始使用

概念

安装与接入

排查问题

CLI 与智能体

支持

用 LogBrew 处理 Flask 日志和 Flask 异常

发送第一个请求

选择你需要的信号

控制日志内容

集成会采集什么

测试异常采集

把出站工作关联到请求

隐私与进程生命周期

排除接入故障

​发送第一个请求

​选择你需要的信号

​控制日志内容

​集成会采集什么

​测试异常采集

​把出站工作关联到请求

​隐私与进程生命周期

​排除接入故障

发送第一个请求

选择你需要的信号

控制日志内容

集成会采集什么

测试异常采集

把出站工作关联到请求

隐私与进程生命周期

排除接入故障