feat(logging): 日志持久化落盘(LOG_FILE 滚动文件) #14

Merged
shangfangjian merged 1 commits from log-persistence into main 2026-09-22 13:34:46 +08:00
11 changed files with 195 additions and 13 deletions
+3
View File
@@ -3,6 +3,9 @@
# production 启动时会强制校验:SECRET_KEY 非空且非弱值、鉴权已配置、DB 弱密码阻断。
APP_ENV=development
LOG_LEVEL=INFO
# 日志持久化:空=仅 stdout + Admin 内存日志页(重启清零);填路径则额外写滚动文件(10MB×5)。
# 本地开发示例: LOG_FILE=./logs/app.log (容器内由 compose 默认设为 /app/logs/app.log 并挂卷)
LOG_FILE=
API_PORT=8000
FRONTEND_PORT=3000
+1
View File
@@ -5,6 +5,7 @@ __pycache__/
.pytest_cache/
frontend/node_modules/
frontend/dist/
logs/
# AI 助手上下文文件(不入库)
CLAUDE.md
+1 -1
View File
@@ -203,7 +203,7 @@ Profeto/
│ │ ├── config.py # pydantic-settings 配置
│ │ ├── crypto.py # 加密/哈希
│ │ ├── http_client.py # 共享 httpx 客户端
│ │ ├── log_buffer.py # 内存日志缓冲(admin 日志页)
│ │ ├── log_buffer.py # 日志基础设施:内存环形缓冲(admin 日志页) + 可选文件持久化(LOG_FILE)
│ │ ├── runtime_config.py # DB 配置覆盖(.env → app_settings)
│ │ ├── scheduler.py # 进程内 cron 调度器
│ │ └── security_check.py # 启动安全校验
+4
View File
@@ -28,6 +28,8 @@ services:
# Fix 2: 容器内 DATABASE_URL 使用 postgres 服务名(非 localhost)
# 必须覆盖 .env 中的 DATABASE_URL,因为 Settings 不读 DB_HOST/DB_PORT
DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER:?POSTGRES_USER 未设置}:${POSTGRES_PASSWORD:?POSTGRES_PASSWORD 未设置}@postgres:5432/${POSTGRES_DB:-football}
# 日志持久化:写入挂载卷,容器重建不丢(应用内滚动 10MB×6 份封顶)
LOG_FILE: /app/logs/app.log
env_file: .env
healthcheck:
test: ["CMD-SHELL", "python -c \"import urllib.request; exit(0 if urllib.request.urlopen('http://localhost:8000/health/ready', timeout=5).status==200 else 1)\" || exit 1"]
@@ -42,6 +44,7 @@ services:
- ./src:/app/src
- ./alembic:/app/alembic
- ./alembic.ini:/app/alembic.ini
- applogs:/app/logs
frontend:
# Fix 4: 多阶段构建 —— 先 build 静态文件,再复制到 nginx
@@ -57,3 +60,4 @@ services:
volumes:
pgdata:
applogs:
+1
View File
@@ -86,6 +86,7 @@ cd frontend && npm install && npm run dev
|---|---|---|---|
| `APP_ENV` | ❌ | `development` | `production` / `development` |
| `LOG_LEVEL` | ❌ | `INFO` | 日志级别 |
| `LOG_FILE` | ❌ | (空) | 日志持久化文件路径;空=仅 stdout + Admin 内存日志页(重启清零)。compose 已默认设为 `/app/logs/app.log` 并挂 `applogs` 卷,滚动上限约 10MB×6 份 |
| `API_PORT` | ❌ | `8000` | API 服务端口映射 |
| `FRONTEND_PORT` | ❌ | `3000` | 前端服务端口映射 |
| `POSTGRES_USER` | ✅ | — | PostgreSQL 用户名 |
+1 -1
View File
@@ -117,7 +117,7 @@ Profeto/
│ ├── config.py # pydantic-settings 配置
│ ├── http_client.py # 共享 httpx 客户端
│ ├── crypto.py # 对称加密(Fernet)与密码哈希
│ ├── log_buffer.py # 内存日志缓冲(admin「系统日志」页)
│ ├── log_buffer.py # 日志基础设施:内存环形缓冲(admin 日志页) + 可选滚动文件持久化(LOG_FILE)
│ ├── runtime_config.py # 运行时配置(数据库优先,回落 .env)
│ ├── scheduler.py # 定时任务调度器(cron 触发采集)
│ └── security_check.py # 生产启动安全校验(缺配置拒绝启动)
+2 -2
View File
@@ -124,8 +124,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
def create_app() -> FastAPI:
from src.core.log_buffer import setup_memory_logging
setup_memory_logging(settings.LOG_LEVEL)
from src.core.log_buffer import setup_logging
setup_logging(settings.LOG_LEVEL, settings.LOG_FILE)
# 生产环境不暴露 OpenAPI 文档(避免向访客泄露接口结构)
openapi_url = "/openapi.json" if settings.APP_ENV != "production" else None
+4
View File
@@ -11,6 +11,10 @@ class Settings(BaseSettings):
# --- app ---
APP_ENV: str = "development"
LOG_LEVEL: str = "INFO"
# 日志持久化:空(默认)只输出 stdout + Admin 内存日志页(重启清零)。
# 填文件路径(如 /app/logs/app.log)后额外写入滚动文件(单文件 10MB × 5 份),
# 进程/容器重启不丢。容器部署需配合 volume 挂载该目录,否则重建仍会丢。
LOG_FILE: str = ""
# P3-3:多 worker 时应用内限流与 KeyRing 各自独立计数(配额放大 N 倍)。
# 设为 True 时若以多 worker 启动 uvicorn 则拒绝启动,避免静默配额漂移。
# 仅在你已前置 Nginx/网关做全局限流、确认不需要此守护时留空/False。
+48 -8
View File
@@ -65,14 +65,54 @@ def get_entries(
return out
def setup_memory_logging(level: str = "INFO") -> None:
"""挂载内存 handler 到 root logger(幂等),并确保 root 级别不低于 INFO。"""
def setup_logging(level: str = "INFO", log_file: str = "") -> None:
"""配置应用日志:stdout(容器收集) + 内存环形缓冲(Admin 日志页) + 可选滚动文件(持久化)。
幂等:重复调用不会重复挂 handler(文件 handler 按 abspath 判重,相对/绝对
路径指向同一文件视为同一个)。文件写入基础设施失败只记 warning,绝不
影响启动与业务 —— 与 _safe_write_ingest_failure 同级约束:可观测性
基础设施不许拖垮主流程。
"""
import os
from logging.handlers import RotatingFileHandler
root = logging.getLogger()
if any(isinstance(h, MemoryLogHandler) for h in root.handlers):
return
handler = MemoryLogHandler()
handler.setLevel(logging.INFO)
handler.addFilter(_SQLNoiseFilter())
root.addHandler(handler)
if root.level == logging.NOTSET or root.level > logging.INFO:
root.setLevel(getattr(logging, level.upper(), logging.INFO))
if not any(isinstance(h, MemoryLogHandler) for h in root.handlers):
handler = MemoryLogHandler()
handler.setLevel(logging.INFO)
handler.addFilter(_SQLNoiseFilter())
root.addHandler(handler)
if not log_file:
return
# 滚动上限:单文件 10MB × 当前+5 份 ≈ 60MB,足够回溯数周的关键事件,
# 又不会吃满磁盘。格式含时间/级别/logger 名,便于事后 grep 排查。
target = os.path.abspath(log_file)
if any(
isinstance(h, RotatingFileHandler) and getattr(h, "baseFilename", None) == target
for h in root.handlers
):
return
try:
os.makedirs(os.path.dirname(target), exist_ok=True)
file_handler = RotatingFileHandler(
target,
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
file_handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
)
file_handler.setLevel(logging.INFO)
file_handler.addFilter(_SQLNoiseFilter())
root.addHandler(file_handler)
logging.getLogger(__name__).info("文件日志已启用: %s", log_file)
except Exception:
logging.getLogger(__name__).warning(
"启用文件日志失败(%s),仅保留 stdout/内存日志", log_file, exc_info=True,
)
+127
View File
@@ -0,0 +1,127 @@
"""日志持久化(LOG_FILE)测试:setup_logging 启用滚动文件后日志必须落盘。
背景: 此前应用日志只进 stdout(docker json-file 收集,不可控) + Admin 内存
日志页(环形缓冲 2000 条,进程重启清零),没有任何应用层持久化 —— 排查
「昨晚采集为什么失败」这类问题时无据可查。本测试守护:
1. 传 log_file → root logger 挂 RotatingFileHandler,日志写入文件
2. 幂等:重复调用不重复挂 handler
3. log_file 为空 → 不挂文件 handler(保持旧行为)
4. 目录不存在 → 自动创建
5. 文件打开失败(如路径是已存在的目录) → 只降级不炸,stdout/内存日志仍在
范式: 直接操作 root logger + tmp_path;fixture 保存/恢复 root 状态,
并关闭新增 handler 的文件句柄(Windows 上句柄不关会锁住 tmp 目录)。
"""
from __future__ import annotations
import logging
from logging.handlers import RotatingFileHandler
import pytest
from src.core.log_buffer import MemoryLogHandler, setup_logging
@pytest.fixture(autouse=True)
def _restore_root_logger():
"""保存/恢复 root logger;关闭测试期间新挂 handler 的句柄。"""
root = logging.getLogger()
saved_handlers = list(root.handlers)
saved_level = root.level
yield
for h in root.handlers:
if h not in saved_handlers and hasattr(h, "close"):
try:
h.close()
except Exception:
pass
root.handlers[:] = saved_handlers
root.setLevel(saved_level)
def _file_handlers():
return [h for h in logging.getLogger().handlers if isinstance(h, RotatingFileHandler)]
def _memory_handlers():
return [h for h in logging.getLogger().handlers if isinstance(h, MemoryLogHandler)]
class TestFileHandlerAttached:
def test_attaches_rotating_file_handler(self, tmp_path):
log_file = tmp_path / "app.log"
setup_logging("INFO", str(log_file))
fhs = _file_handlers()
assert len(fhs) == 1
assert fhs[0].baseFilename == str(log_file)
# 内存日志页照常工作,两者并存
assert len(_memory_handlers()) == 1
# 滚动参数与文档口径一致: 单文件 10MB × 5 份
assert fhs[0].maxBytes == 10 * 1024 * 1024
assert fhs[0].backupCount == 5
def test_log_written_to_file(self, tmp_path):
log_file = tmp_path / "app.log"
setup_logging("INFO", str(log_file))
logging.getLogger("persist-test").info("hello-persist-12345")
content = log_file.read_text(encoding="utf-8")
assert "hello-persist-12345" in content
assert "persist-test" in content # logger 名可追溯
assert "INFO" in content # 级别在行首可过滤
class TestIdempotent:
def test_repeated_call_does_not_duplicate_handlers(self, tmp_path):
log_file = tmp_path / "app.log"
setup_logging("INFO", str(log_file))
setup_logging("INFO", str(log_file))
setup_logging("INFO", str(log_file))
assert len(_file_handlers()) == 1
assert len(_memory_handlers()) == 1
def test_same_file_via_relative_and_absolute_path_counts_as_one(self, tmp_path, monkeypatch):
"""相对/绝对路径指向同一文件时不得重复挂(幂等按 abspath 判重)。"""
monkeypatch.chdir(tmp_path)
setup_logging("INFO", "app.log")
setup_logging("INFO", str(tmp_path / "app.log"))
assert len(_file_handlers()) == 1
class TestDisabled:
def test_empty_log_file_keeps_old_behavior(self):
setup_logging("INFO", "")
assert _file_handlers() == []
assert len(_memory_handlers()) == 1
class TestAutoMkdir:
def test_creates_missing_directories(self, tmp_path):
log_file = tmp_path / "deep" / "nested" / "app.log"
setup_logging("INFO", str(log_file))
assert len(_file_handlers()) == 1
assert log_file.parent.is_dir()
class TestGracefulDegradation:
def test_unopenable_path_degrades_without_raising(self, tmp_path):
"""路径是已存在的目录 → 打开必然失败;只降级,不炸启动。"""
dir_as_file = tmp_path / "occupied"
dir_as_file.mkdir()
# 不应抛异常(文件日志是可观测性基础设施,失败只 warning)
setup_logging("INFO", str(dir_as_file))
assert _file_handlers() == []
# 降级后 stdout/内存日志路径仍在
assert len(_memory_handlers()) == 1
+3 -1
View File
@@ -73,7 +73,9 @@ class TestStandingsModel:
def test_unique_constraint_includes_available_at(self):
names = {c.name for c in Standing.__table__.constraints}
assert any("available" in n and n.startswith("uq_") for n in names), \
# 匿名约束(PK/Check)的 name 是 None,必须先判真值再子串匹配,
# 否则 any() 撞上 None 直接 TypeError(集合顺序不定 → flaky)
assert any(n and "available" in n and n.startswith("uq_") for n in names), \
f"缺少含 available_at 的唯一约束,现有: {names}"
def test_old_unique_constraint_removed(self):