From cdcfcb2e4313f3bbf640bca241cb0edcac5c661b Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 21 Aug 2026 13:47:44 +0800 Subject: [PATCH] feat(audit): skip audit log for excluded health/root paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 健康检查与根路径（/health/live、/health/ready、/api/v1/health、 /、/health/storage）没有用户、没业务动作，每秒被 K8s/LB 探针刷一次只会灌进无意义噪音。命中排除集即跳过审计行； 诊断日志（method/path/status/ms 走 stderr）照常打，对容器 运维排错仍有用。 * settings.audit_excluded_paths: list[str] 默认覆盖 5 条 基础设施路径，env AUDIT_EXCLUDED_PATHS 用逗号分隔 （pydantic NoDecode + field_validator 兼容 str/list） * main.py 模块级 _AUDIT_EXCLUDED = frozenset(...)， access_log 的 success/exception 两条审计行各加守卫 诊断无条件打 * 测试用 _AccessLogReplica 复刻 access_log 契约（不 import 真实 main.py），新增 4 个 case：排除根路径、排除 /health/live、 不排除路径照写审计、自定义排除集 顺带 schedule 模块：ExecutionResult 与 context 已迁到 schedule.domain.*（execution.py / orchestrator.py / scheduler.py / worker.py），调用点跟进；schedule 自身 18 个测试在改前改后均通过。 --- .env.example | 5 ++ backend/src/backend/main.py | 34 +++++--- backend/tests/test_audit_logging.py | 117 +++++++++++++++++++++++--- common/src/common/config.py | 28 +++++- schedule/src/schedule/execution.py | 14 +-- schedule/src/schedule/orchestrator.py | 2 +- schedule/src/schedule/scheduler.py | 2 +- schedule/src/schedule/worker.py | 5 +- 8 files changed, 162 insertions(+), 45 deletions(-) diff --git a/.env.example b/.env.example index d1527b7..165c225 100644 --- a/.env.example +++ b/.env.example @@ -36,6 +36,11 @@ LOG_LEVEL=INFO # container). AUDIT_LOG_RETENTION_DAYS=0 disables cleanup of old files. AUDIT_LOG_DIR= AUDIT_LOG_RETENTION_DAYS=30 +# Exact paths excluded from the audit line (health/root probes carry no +# business value but fire every second from K8s/LB). The default already +# covers /health/live /health/ready /api/v1/health / /health/storage; +# leave empty to keep the default. Comma-separated, e.g. /health/live,/api/v1/health. +AUDIT_EXCLUDED_PATHS= # Object storage. Two modes are supported: # STORAGE_BACKEND=s3 — connects to an S3-compatible service (MinIO, diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 4fea196..af2f776 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -53,6 +53,12 @@ configure_audit_logging( ) +# 审计排除的精确路径集（不含 query）：命中即跳过审计行，诊断日志照常打。 +# 健康检查 / 根路径探针每秒刷审计文件但无业务价值；可用 +# settings.audit_excluded_paths / AUDIT_EXCLUDED_PATHS 覆盖默认值。 +_AUDIT_EXCLUDED: frozenset[str] = frozenset(settings.audit_excluded_paths) + + @asynccontextmanager async def lifespan(app: Any) -> AsyncIterator[None]: # 生命周期内创建的对象挂在 app.state 上，路由通过 Depends 或 Request @@ -137,7 +143,9 @@ async def access_log(request: Request, call_next): # 诊断：方法/路径/状态码/耗时 走 stderr（loguru default sink） # 合规：时间/用户/方法/路径/状态码 走独立 audit 文件 sink # 两条 logger.info() 共用一个出口，便于排查 + # 排除集是精确路径匹配（不含 query），命中即跳过审计行；诊断日志照常。 start = time.perf_counter() + skip_audit = request.url.path in _AUDIT_EXCLUDED try: response = await call_next(request) except Exception: @@ -147,12 +155,13 @@ async def access_log(request: Request, call_next): method=request.method, path=request.url.path, ms=elapsed_ms, ) # 异常路径：审计行也要写（status=500 由 unhandled_exception_handler 返回） - logger.bind( - user_id=_audit_user_id(request), - method=request.method, - path=request.url.path, - status=500, - ).info("audit") + if not skip_audit: + logger.bind( + user_id=_audit_user_id(request), + method=request.method, + path=request.url.path, + status=500, + ).info("audit") raise elapsed_ms = (time.perf_counter() - start) * 1000 logger.info( @@ -160,12 +169,13 @@ async def access_log(request: Request, call_next): method=request.method, path=request.url.path, status=response.status_code, ms=elapsed_ms, ) - logger.bind( - user_id=_audit_user_id(request), - method=request.method, - path=request.url.path, - status=response.status_code, - ).info("audit") + if not skip_audit: + logger.bind( + user_id=_audit_user_id(request), + method=request.method, + path=request.url.path, + status=response.status_code, + ).info("audit") return response diff --git a/backend/tests/test_audit_logging.py b/backend/tests/test_audit_logging.py index f767971..e2dca17 100644 --- a/backend/tests/test_audit_logging.py +++ b/backend/tests/test_audit_logging.py @@ -53,29 +53,38 @@ class _AccessLogReplica(BaseHTTPMiddleware): """复制 main.py access_log 写审计行的契约（不 import 真实 main.py）。 测试只覆盖 access_log 的审计行为（success / exception 两条路径 + - user_id 解析），避免触发 main.py 的 lifespan（MySQL / 路由初始化）。 - 未来 main.py 改 access_log 字段时，这里同步改即可，测试不会假阳/假阴。 + user_id 解析 + 排除集守卫），避免触发 main.py 的 lifespan（MySQL / + 路由初始化）。``excluded_paths`` 对应 main.py 的 ``_AUDIT_EXCLUDED``： + 精确路径命中即跳过审计行（诊断 stderr 照常）。未来 main.py 改 + access_log 字段时，这里同步改即可，测试不会假阳/假阴。 """ + def __init__(self, app, excluded_paths: frozenset[str] = frozenset()): + super().__init__(app) + self.excluded_paths = excluded_paths + async def dispatch( self, request: Request, call_next: Callable[..., Awaitable[Response]] ) -> Response: + skip_audit = request.url.path in self.excluded_paths try: response = await call_next(request) except Exception: + if not skip_audit: + logger.bind( + user_id=_audit_user_id(request), + method=request.method, + path=request.url.path, + status=500, + ).info("audit") + raise + if not skip_audit: logger.bind( user_id=_audit_user_id(request), method=request.method, path=request.url.path, - status=500, + status=response.status_code, ).info("audit") - raise - logger.bind( - user_id=_audit_user_id(request), - method=request.method, - path=request.url.path, - status=response.status_code, - ).info("audit") return response @@ -95,12 +104,17 @@ def _reset_audit_logging(): def _build_client( - log_dir: str, *, raise_server_exceptions: bool = True + log_dir: str, + *, + raise_server_exceptions: bool = True, + excluded_paths: frozenset[str] = frozenset(), ) -> TestClient: """配置审计日志并返回挂上 access_log 契约复刻中间件的测试 app 客户端。 - 只注册空路由，不碰数据库与业务逻辑；中间件在路由之后注册，与 main.py - 的 access_log 行为一致。``/boom`` 用于验证 exception 路径（500）。 + 只注册少量测试路由，不碰数据库与业务逻辑；中间件在路由之后注册，与 + main.py 的 access_log 行为一致。``/boom`` 用于验证 exception 路径 + （500）；``/health/live`` ``/api/v1/scripts`` ``/custom`` ``/other`` + ``/`` 供排除集用例使用。 """ audit.configure_audit_logging(log_dir, retention_days=30) @@ -114,7 +128,27 @@ def _build_client( def boom() -> dict: raise RuntimeError("boom") - app.add_middleware(_AccessLogReplica) + @app.get("/health/live") + def health_live() -> dict: + return {"ok": True} + + @app.get("/api/v1/scripts") + def scripts() -> dict: + return {"ok": True} + + @app.get("/custom") + def custom() -> dict: + return {"ok": True} + + @app.get("/other") + def other() -> dict: + return {"ok": True} + + @app.get("/") + def root() -> dict: + return {"ok": True} + + app.add_middleware(_AccessLogReplica, excluded_paths=excluded_paths) return TestClient(app, raise_server_exceptions=raise_server_exceptions) @@ -206,6 +240,61 @@ def test_audit_log_bearer_header_resolves_user(tmp_path: Path) -> None: assert f"| {user_id} | GET /x -> 200" in lines[0] +def test_access_log_skips_audit_for_excluded_path(tmp_path: Path) -> None: + """命中排除集（/health/live）：审计行不写，诊断 stderr 照常。 + + stderr 走 loguru default sink，难以用 caplog 抓取，这里直接断言 + 审计文件无新行即可。 + """ + client = _build_client( + str(tmp_path), excluded_paths=frozenset({"/health/live"}) + ) + response = client.get("/health/live") + assert response.status_code == 200 + + assert _audit_lines(tmp_path) == [] + + +def test_access_log_skips_audit_for_root_path(tmp_path: Path) -> None: + """根路径 `/` 命中排除集：审计行不写。""" + client = _build_client(str(tmp_path), excluded_paths=frozenset({"/"})) + response = client.get("/") + assert response.status_code == 200 + + assert _audit_lines(tmp_path) == [] + + +def test_access_log_still_writes_audit_for_non_excluded_path( + tmp_path: Path, +) -> None: + """未排除路径（/api/v1/scripts）照常写审计行。""" + client = _build_client( + str(tmp_path), excluded_paths=frozenset({"/health/live", "/"}) + ) + response = client.get("/api/v1/scripts") + assert response.status_code == 200 + + lines = _audit_lines(tmp_path) + assert lines + assert "GET /api/v1/scripts -> 200" in lines[0] + + +def test_access_log_custom_excluded_paths_from_settings(tmp_path: Path) -> None: + """自定义排除集：/custom 命中跳过、/other 未命中照常写。""" + client = _build_client( + str(tmp_path), excluded_paths=frozenset({"/custom"}) + ) + response = client.get("/custom") + assert response.status_code == 200 + assert _audit_lines(tmp_path) == [] + + response = client.get("/other") + assert response.status_code == 200 + lines = _audit_lines(tmp_path) + assert lines + assert "GET /other -> 200" in lines[0] + + def test_retention_cleanup_removes_old_files(tmp_path: Path) -> None: old = tmp_path / "audit-2024-01-01.log" recent = tmp_path / "audit-2024-06-01.log" diff --git a/common/src/common/config.py b/common/src/common/config.py index 8fc9394..cf471a2 100644 --- a/common/src/common/config.py +++ b/common/src/common/config.py @@ -18,9 +18,10 @@ Rules for adding a new variable: from __future__ import annotations from functools import lru_cache +from typing import Annotated -from pydantic import Field -from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict class Settings(BaseSettings): @@ -79,6 +80,29 @@ class Settings(BaseSettings): default=30, description="审计日志保留天数；过期文件启动时清理。设 0 关闭清理。", ) + audit_excluded_paths: Annotated[list[str], NoDecode] = Field( + default=[ + "/health/live", + "/health/ready", + "/api/v1/health", + "/", + "/health/storage", + ], + description=( + "审计排除的精确路径列表（不含 query）。命中即不写审计行。" + "诊断日志（method/path/status/ms）仍写 stderr。" + "环境变量 AUDIT_EXCLUDED_PATHS 用逗号分隔，例如" + " '/health/live,/api/v1/health'。" + ), + ) + + @field_validator("audit_excluded_paths", mode="before") + @classmethod + def _split_audit_paths(cls, v): + # env 进来是 "a,b,c"；代码里直接传 list 也行 + if isinstance(v, str): + return [s.strip() for s in v.split(",") if s.strip()] + return v # ── runtime container endpoint ─────────────────────────────── runtime_api_url: str = Field( diff --git a/schedule/src/schedule/execution.py b/schedule/src/schedule/execution.py index deae579..09ce315 100644 --- a/schedule/src/schedule/execution.py +++ b/schedule/src/schedule/execution.py @@ -4,10 +4,10 @@ import asyncio import json import sys import tempfile -from dataclasses import dataclass from pathlib import Path, PurePosixPath from loguru import logger +from schedule.domain.execution import ExecutionResult MAX_LOG_BYTES = 4 * 1024 * 1024 @@ -26,18 +26,6 @@ def _resolve_python(version: str) -> str: raise ValueError(f"unsupported python_version: {version}") -@dataclass(frozen=True) -class ExecutionResult: - status: str - exit_code: int | None - logs: bytes - result: bytes - result_file_name: str - result_content_type: str - error_code: str | None = None - error_message: str | None = None - - def _limited_log(value: str) -> bytes: encoded = value.encode("utf-8", errors="replace") if len(encoded) <= MAX_LOG_BYTES: diff --git a/schedule/src/schedule/orchestrator.py b/schedule/src/schedule/orchestrator.py index 48034dc..69e921a 100644 --- a/schedule/src/schedule/orchestrator.py +++ b/schedule/src/schedule/orchestrator.py @@ -30,7 +30,7 @@ from loguru import logger from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker -from schedule.context import ( +from schedule.domain.context import ( FAILED_NODE_STATES, TERMINAL_NODE_STATES, TERMINAL_RUN_STATES, diff --git a/schedule/src/schedule/scheduler.py b/schedule/src/schedule/scheduler.py index 5de0167..bf57dc2 100644 --- a/schedule/src/schedule/scheduler.py +++ b/schedule/src/schedule/scheduler.py @@ -28,7 +28,7 @@ from loguru import logger from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker -from schedule.context import naive_utc +from schedule.domain.context import naive_utc # APScheduler's persistent SQLAlchemy job store pickles each job. A bound # ``CronScheduler`` method captures this instance (including SQLAlchemy engine diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py index 436d148..f000c1d 100644 --- a/schedule/src/schedule/worker.py +++ b/schedule/src/schedule/worker.py @@ -41,8 +41,9 @@ from common.scheduler.trigger import SYSTEM_CRON_USER_ID from loguru import logger from sqlalchemy import select -from schedule.context import TERMINAL_NODE_STATES -from schedule.execution import ExecutionResult, execute_artifact +from schedule.domain.context import TERMINAL_NODE_STATES +from schedule.domain.execution import ExecutionResult +from schedule.execution import execute_artifact NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute") NODE_FINISHED_EVENT = schedule_event_type("job.node.finished")