---
name: observability-rules
description: "系统可观测性与日志级别规范。适用于所有系统组件日志记录、指标上报与链路追踪 TraceID 注入。"
globs: ["backend/**/*.py", "src/**/*.ts", "**/logger.ts", "**/*.py", "**/*.ts"]
alwaysApply: false
updated: 2026-05-22
---

# 系统可观测性与日志级别规范

> [!IMPORTANT]
> 优秀的系统可观测性是在黑暗中穿行的明灯。AI 助手在编写代码或插入日志记录点时，必须严格区分日志等级，并提供结构化、可跨链路追溯的日志载荷。

## 1. 适用场景

当您在控制器处理请求流、捕获内部异常、调用外部接口服务，或在底层核心逻辑中插入状态输出记录时。

## 2. 操作规则

1. **精准分级日志级别 (Strict Log Leveling)**：
   - **`DEBUG`**：记录底层详细的二进制流、复杂 SQL 转换等，生产环境默认关闭。
   - **`INFO`**：记录核心生命周期的正向推进节点（如“安全支付已成功完成”、“用户登录会话建立”）。
   - **`WARN`**：记录可自动容错修复的非致命异常（如“第三方 API 首次超时，触发第一次自动重试”、“发现过期 Token 自动重定向”）。
   - **`ERROR`**：记录阻断单次业务正常进行的失败故障（如“外部网关扣减余额超时”、“数据库存储报错”），必须携带完整异常堆栈。
2. **结构化 JSON 格式输出 (Structured Logging)**：
   - 生产环境输出日志**禁止**直接打印裸文本字符串。必须统一采用 **JSON 结构化流** 格式，将额外上下文参数作为独立字典字段（如 `extra={"user_id": 123}`）传入，利于 ELK/Loki 索引。
3. **TraceID 分布式链路穿透 (Distributed TraceID Injection)**：
   - 所有在 API 控制器或消费队列中生成的日志，必须在最外层提取或生成统一的链路标识（`trace_id`）。
   - 该 `trace_id` 必须贯穿并在该请求引发的所有数据库操作、外部微服务 RPC 请求的 HTTP Header 中进行穿透传递，以便于通过单一 TraceID 秒级还原完整请求调用树。

## 3. 禁止事项

- 严禁在日志中记录任何涉及用户隐私的脱敏前敏感参数（如明文密码、银行卡密）。

## 4. 验证方式

- 运行业务请求，观察控制台输出日志的 JSON 结构中是否均带有 `level`、`trace_id` 和 `message`。

## 5. 代码对比示例

### ❌ 错误示例（非结构化裸文本，无法跨系统定位链路，日志等级滥用）

```python
# 违反规则 2.2：打印无结构的纯字符串，机器极难过滤索引
# 违反规则 2.1：仅仅是一个普通的重试非崩溃场景，滥用了高警报级别的 ERROR
# 违反规则 2.3：缺少 trace_id，在微服务中根本找不到是哪个 HTTP 请求触发的日志
def call_external_gateway_bad(user_id):
    try:
        response = gateway.charge(user_id, amount=100)
    except TimeoutException as e:
        logger.error(f"FATAL SYSTEM FAILURE!!! CHARGING USER {user_id} FAILED DUE TO TIMEOUT EXCEPTION!!!")
        return retry_charge(user_id)
```

###  正确方向（结构化 JSON 字段、Warn级自动重试、TraceID 分布式链路传递）

```python
import uuid

# 遵循规则 2.2 & 2.3：结构化 JSON 日志配置，并提取当前上下文的 trace_id 穿透定位
def call_external_gateway_optimized(user_id, context_trace_id=None):
    # 如果没有传递链路 TraceID，则自建以防止链路中断
    trace_id = context_trace_id or str(uuid.uuid4())
    
    log_context = {
        "trace_id": trace_id,
        "user_id": user_id,
        "action": "charge_gateway_call"
    }

    try:
        # 遵循规则 2.1：记录正向核心节点的 INFO 日志
        logger.info("Initiating billing gateway charge call", extra=log_context)
        response = gateway.charge(user_id, amount=100)
        return response
    except TimeoutException as exc:
        # 遵循规则 2.1：因为此处会自动触发容错重试，业务未彻底失败崩溃，应当精准分级为 WARNING 告警
        logger.warning(
            "Gateway timeout encountered. Initiating secondary retry attempt.",
            exc_info=exc,
            extra=log_context
        )
        return retry_charge_with_trace(user_id, trace_id)
```