# 生产级日志系统设计方案

> 版本：v1.0  
> 日期：2026-08-17  
> 状态：待评审

---

## 一、现状分析

### 1.1 当前日志架构

```
┌─────────────┐
│  FastAPI    │  ← 请求入口
└──────┬──────┘
       │
       ▼
┌─────────────┐
│ Logging     │  ← Python logging 模块
│ Middleware   │
└──────┬──────┘
       │
       ├──────────────────┬──────────────────┐
       ▼                  ▼
┌─────────────┐      ┌─────────────┐
│ Console    │      │ File        │  ← 输出目标
│ (开发环境)  │      │ (JSON格式)   │
└─────────────┘      └─────────────┘
                            │
                            ▼
                     ┌─────────────┐
                     │ audit_logs  │  ← 审计日志表
                     │ (PostgreSQL) │
                     └─────────────┘
```

### 1.2 现有组件

| 组件 | 文件路径 | 功能 |
|------|----------|------|
| `logging.py` | `backend/app/core/` | 日志配置、JSON 格式化、按日期/大小轮转 |
| `platform_store.py` | `backend/app/db/` | `record_audit()` 审计日志写入 |
| `002_governance.sql` | `backend/app/db/sql/` | `audit_logs` 表结构 |

### 1.3 存在的问题

| 问题 | 影响 | 严重程度 |
|------|------|----------|
| **无结构化日志分级** | DEBUG/INFO/WARNING/ERROR 全部混在一起，无法按级别过滤查看 | 🔴 高 |
| **无请求链路追踪** | 一个请求从进入到返回经过哪些服务/函数，无法串联 | 🔴 高 |
| **审计日志与业务耦合** | 各模块手动调用 `record_audit()`，容易遗漏 | 🟡 中 |
| **无敏感数据脱敏** | 用户 token、密码等可能明文记录 | 🔴 高 |
| **无日志聚合查询** | 无法按用户/时间范围/操作类型快速检索 | 🟡 中 |
| **无告警通知** | 系统异常无法主动推送通知 | 🟡 中 |
| **日志文件无归档策略** | 只有简单的过期删除，无压缩归档 | 🟢 低 |

---

## 二、设计目标

### 2.1 核心原则

1. **结构化** - 日志有固定 schema，便于机器解析和查询
2. **可追溯** - 每个请求有唯一 ID，可串联完整调用链路
3. **分级输出** - 不同环境输出不同级别，生产环境不输出 DEBUG
4. **安全合规** - 敏感数据自动脱敏（token、密码、手机号等）
5. **高性能** - 日志写入不影响业务接口性能（异步写入）
6. **可观测** - 支持快速检索、统计、告警

### 2.2 日志分级标准

| 级别 | 使用场景 | 示例 | 生产环境 |
|------|----------|------|:--------:|
| **DEBUG** | 开发调试 | 变量值、SQL 语句、完整堆栈 | ❌ 不输出 |
| **INFO** | 正常流程记录 | 任务创建成功、用户登录 | ✅ 记录 |
| **WARNING** | 可恢复异常 | 重试操作、参数校验失败、资源不足 | ✅ 记录 |
| **ERROR** | 需要人工介入 | 数据库连接失败、第三方 API 超时 | ✅ 记录 + 告警 |
| **CRITICAL** | 系统不可用 | 磁盘满、主节点宕机 | ✅ 记录 + 立即告警 |

---

## 三、技术方案

### 3.1 整体架构

```
┌─────────────────────────────────────────────────────────────────────┐
│                        应用层 (Application Layer)                      │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐         │
│  │ 数据集管理 │  │ 微调训练  │  │ 模型推理  │  │ 用户认证  │  ...     │
│  └─────┬────┘  └─────┬────┘  └─────┬────┘  └─────┬────┘         │
│        │            │           │          │                 │
│        └────────────┴───────────┴──────────┘                 │
│                             ▼                             │
│                    ┌──────────────┐                       │
│                    │ Structured   │  ← 结构化日志中间件     │
│                    │ Logger       │                       │
│                    └──────┬───────┘                       │
│                           │                              │
│              ┌────────────┬────────────┬─────────────┐       │
│              ▼            ▼            ▼             │       │
│     ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│     │ Console  │  │ File     │  │ 审计DB   │  │ 告警   │ │
│     │ (开发)   │  │ (JSON)   │  │ (PG)     │  │(可选)  │ │
│     └──────────┘  └──────────┘  └──────────┘  └────────┘ │
│                                                     │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                        可观测层 (Observability)                   │
├─────────────────────────────────────────────────────────────┤
│  ┌───────────┐  ┌───────────┐  ┌───────────┐              │
│  │ Grafana   │  │ Kibana   │  │ PagerDuty │  ...         │
│  │ (查询)    │  │ (分析)   │  │ (告警)   │              │
│  └───────────┘  └───────────┘  └───────────┘              │
└─────────────────────────────────────────────────────────────┘
```

### 3.2 日志 Schema 设计

#### 3.2.1 应用日志 (app.log)

```json
{
  "timestamp": "2026-08-17T10:30:00.000Z",
  "level": "INFO",
  "trace_id": "req-abc123",
  "parent_span_id": "span-xyz789",  // OpenTelemetry Span
  "request": {
    "method": "POST",
    "path": "/dataset-manage",
    "client_ip": "192.168.1.100",
    "user_agent": "Mozilla/5.0...",
    "user_id": "u_admin"
  },
  "module": "dataset.router",
  "function": "create_dataset",
  "message": "数据集创建成功",
  "extra": {
    "dataset_id": "ds_abc123",
    "dataset_name": "训练数据"
  },
  "duration_ms": 125,
  "status_code": 200,
  "error": null
}
```

#### 3.2.2 审计日志 (audit_logs 表)

```sql
-- 已有表结构（保持不变）
CREATE TABLE IF NOT EXISTS audit_logs (
  id TEXT PRIMARY KEY,
  tenant_id TEXT,
  project_id TEXT,
  actor_id TEXT,           -- 操作人
  action TEXT,               -- 操作类型: create/delete/update/acl.set/login...
  target_type TEXT,          -- 资源类型: dataset/model/fine-tune/user...
  target_id TEXT,            -- 资源 ID
  detail TEXT,               -- 详细信息 JSON
  client_ip TEXT,            -- 客户端 IP
  time TEXT,                -- 操作时间
  
  -- 新增字段
  trace_id TEXT,            -- 关联应用日志的请求追踪 ID
  request_method TEXT,      -- HTTP 方法
  request_path TEXT,        -- 请求路径
  status_code INTEGER,       -- 响应状态码
  duration_ms REAL,          -- 耗时(ms)
  extra JSONB               -- 扩展信息
);

-- 新增索引
CREATE INDEX IF NOT EXISTS idx_audit_trace ON audit_logs(trace_id);
CREATE INDEX IF NOT EXISTS idx_audit_actor_time ON audit_logs(actor_id, time);
```

### 3.3 日志中间件设计

```python
# backend/app/core/logging.py 新增

class StructuredLogger:
    """结构化日志记录器"""
    
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.trace_id = context_var.get("trace_id")
    
    def info(self, msg: str, **kwargs):
        self._log("INFO", msg, **kwargs)
    
    def warning(self, msg: str, **kwargs):
        self._log("WARNING", msg, **kwargs)
    
    def error(self, msg: str, **kwargs):
        self._log("ERROR", msg, **kwargs)
    
    def _log(self, level: str, msg: str, 
         user_id: str = None,
         target_type: str = None,
         target_id: str = None,
         duration_ms: float = None,
         status_code: int = None,
         error: Exception = None,
         **extra):
        """统一日志记录方法"""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": level,
            "trace_id": self.trace_id.get(),
            "request": {
                "user_id": user_id or current_user_id(),
                "client_ip": client_ip(),
                # ...
            },
            "module": calling_module,
            "message": msg,
            "target": {
                "type": target_type,
                "id": target_id,
            },
            "extra": extra,
            "duration_ms": duration_ms,
            "error": format_exception(error) if error else None,
        }
        
        # 1. 写入控制台/文件
        self.logger.log(level, json.dumps(log_entry))
        
        # 2. 异步写入审计表（如果需要）
        if level in ("WARNING", "ERROR", "CRITICAL"):
            async_write_audit(log_entry)
```

### 3.4 装饰器模式（推荐）

使用 Python 裁饰器自动记录，避免手动调用：

```python
# backend/app/core/log_decorator.py

def audit_log(action: str, target_type: str = ""):
    """审计日志装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            result = await func(*args, **kwargs)
            
            # 自动记录审计日志
            record_audit(
                action=action,
                target_type=target_type,
                target_id=kwargs.get('id') or result.get('id'),
                detail=f"params={kwargs}"
            )
            return result
        return wrapper
    return decorator


# 使用示例
@audit_log("dataset.create", "dataset")
async def create_dataset(...):
    # 业务逻辑
    pass
```

### 3.5 敏感数据脱敏规则

```python
# backend/app/core/masking.py

SENSITIVE_FIELDS = {
    "token": "***",
    "password": "***",
    "phone": lambda x: f"{x[:3]}****{x[-4:]}",
    "email": lambda x: x[0] + "***" + x.split("@")[1] if "@" in x else "***",
    "id_card": lambda x: f"{x[:6]}********{x[-4:]}",
}

def mask_sensitive(data: dict) -> dict:
    """递归脱敏字典中的敏感字段"""
    for key, value in data.items():
        if key in SENSITIVE_FIELDS:
            data[key] = SENSITIVE_FIELDS[key](value) if callable(SENSITIVE_FIELDS[key]) else "***"
        elif isinstance(value, dict):
            mask_sensitive(value)
    return data
```

---

## 四、实施计划

### 4.1 Phase 1：基础增强（1-2 天）

- [ ] **P1-1** 升级 `JsonLogFormatter`，增加 `trace_id` 字段
- [ ] **P1-2** 新增 `StructuredLogger` 封装类
- [ ] **P1-3** 统一所有模块的日志格式为 JSON
- [ ] **P1-4** 实现 `mask_sensitive()` 脱敏函数
- [ ] **P1-5** 审计日志表新增 `trace_id`、`duration_ms` 字段

### 4.2 Phase 2：自动化（2-3 天）

- [ ] **P2-1** 编写 `@audit_log` 装饰器
- [ ] **P2-2** 为关键业务接口添加装饰器：
  - 数据集 CRUD
  - 模型 CRUD
  - 微调任务创建/删除
  - 用户登录/登出
  - ACL 授权变更
- [ ] **P2-3** 实现日志异步写入队列（避免影响性能）

### 4.3 Phase 3：可观测性（3-5 天）

- [ ] **P3-1** 集成 ELK Stack 或 Loki（可选）
- [ ] **P3-2** 编写 Grafana 仪表板：
  - 请求量趋势图
  - 错误率统计
  - 慢接口 TOP10
  - 用户操作审计面板
- [ ] [ ] **P3-3** 实现告警规则（错误率超阈值触发）

---

## 五、配置示例

### 5.1 日志配置 (settings)

```yaml
# config.yaml 或 .env
LOGGING:
  level: INFO          # 生产环境用 INFO，开发用 DEBUG
  dir: ./logs
  file_prefix: app
  max_bytes: 50MB       # 单文件最大 50MB
  retention_days: 30      # 保留 30 天
  error_prefix: error   # 错误日志单独文件
  json: true            # JSON 格式输出

AUDIT:
  enabled: true
  auto_record: true     # 是否自动记录（通过装饰器）
  sensitive_mask: true   # 启用敏感数据脱敏
```

### 5.2 日志输出示例

**控制台输出（开发环境）：**
```
2026-08-17 18:30:00.123 | INFO | pid=12345 | MainThread | req=req-abc | dataset.router:create_dataset | dataset/router.py:45 | 数据集创建成功 {"dataset_id":"ds_abc"}
```

**文件输出（JSON 格式）：**
```json
{"@timestamp":"2026-08-17T18:30:00.123Z","level":"INFO","logger":"dataset.router","message":"数据集创建成功","module":"dataset.router","function":"create_dataset","file":"dataset/router.py","line":45,"process":12345,"thread":"MainThread","request_id":"req-abc","extra":{"dataset_id":"ds_abc"}}
```

**审计日志查询 SQL：**
```sql
-- 查询某用户最近7天的所有操作
SELECT time, action, target_type, target_id, detail, client_ip
FROM audit_logs
WHERE actor_id = 'u_admin'
  AND time >= now() - interval '7 days'
ORDER BY time DESC;

-- 查询某资源的授权变更历史
SELECT * FROM audit_logs
WHERE action LIKE '%acl%'
  AND target_id = 'ds_abc123'
ORDER BY time DESC;
```

---

## 六、附录

### A. 日志关键字段说明

| 字段 | 类型 | 说明 | 示例 |
|------|------|------|------|
| `trace_id` | string | 请求唯一标识，用于串联一次请求的所有日志 | `req-uuid-1234` |
| `parent_span_id` | string | 父 Span ID（用于分布式追踪） | `span-parent-5678` |
| `actor_id` | string | 操作人用户 ID | `u_admin` |
| `action` | string | 操作动作 | `dataset.create`, `model.delete`, `login.success` |
| `target_type` | string | 操作的资源类型 | `dataset`, `trained_model`, `user` |
| `target_id` | string | 资源 ID | `ds_abc123` |
| `detail` | string/json | 操作详情 | `{"name": "训练数据", "type": "train"}` |
| `client_ip` | string | 客户端 IP | `192.168.1.100` |
| `duration_ms` | real | 接口耗时(ms) | `125.5` |
| `status_code` | int | HTTP 状态码 | `200`, `404`, `500` |

### B. 推荐的 Python 日志库对比

| 库 | 特点 | 适用场景 |
|-----|------|---------|
| `structlog` | 结构化日志，高性能 | 推荐 ✅ |
| `loguru` | 简单易用，自动配置 | 小型项目 |
| `logging` | Python 标准库 | 当前已使用 |

### C. 参考链接

- [Python logging cookbook](https://docs.python.org/3/howto/logging.html)
- [ELK Stack 官方文档](https://www.elastic.co/guide/index.html)
- [OpenTelemetry 规范](https://opentelemetry.io/docs/)
