diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py
index b706779..2d7b657 100644
--- a/backend/app/api/v1/endpoints/data_process.py
+++ b/backend/app/api/v1/endpoints/data_process.py
@@ -35,6 +35,7 @@ from fastapi.responses import StreamingResponse
from psycopg.rows import dict_row
from app.core.auth import filter_accessible_resource_ids, get_current_user, is_admin
+from app.core.logging import get_structured_logger
from app.modules.data_process.algorithms import (
ParsedText,
canonical_record_json,
@@ -103,6 +104,7 @@ from app.schemas.data_process import (
router = APIRouter(prefix="/data-process")
logger = logging.getLogger(__name__)
+biz_logger = get_structured_logger("app.biz.data_process")
MAX_SOURCE_FILE_BYTES = 200 * 1024 * 1024
MAX_SOURCE_FILE_COUNT = 20
MAX_SOURCE_BATCH_BYTES = 500 * 1024 * 1024
@@ -854,6 +856,7 @@ def create_task(
) -> dict[str, Any]:
with api_errors():
task = store.create_task(payload.model_dump(mode="json"))
+ biz_logger.info("data-process:create success", taskId=task["id"], processType=task.get("process_type", ""))
return ok(task, "data process task created")
@@ -879,10 +882,9 @@ def update_task(
store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
with api_errors():
- return ok(
- store.update_task(task_id, payload.model_dump(exclude_unset=True, mode="json")),
- "data process task updated",
- )
+ result = store.update_task(task_id, payload.model_dump(exclude_unset=True, mode="json"))
+ biz_logger.info("data-process:update success", taskId=task_id)
+ return ok(result, "data process task updated")
@router.put("/{task_id}/workflow-step")
@@ -1047,6 +1049,7 @@ def delete_task(
) -> dict[str, Any]:
with api_errors():
store.delete_task(task_id)
+ biz_logger.info("data-process:delete success", taskId=task_id)
return ok({"deleted": task_id}, "data process task deleted")
@@ -1359,13 +1362,12 @@ def delete_source_file(
expected_source_file_id=file_id,
)
except Exception:
- # 数据库软删除已经提交,不能再向客户端返回可重试的失败;保留逻辑引用,
- # 由后续存储清理任务重试物理删除。
cleanup_pending = True
logger.exception(
"failed to remove data process source object after soft deletion",
extra={"task_id": task_id, "source_file_id": file_id},
)
+ biz_logger.info("data-process:delete-source-file success", taskId=task_id, fileId=file_id, storageCleanupPending=cleanup_pending)
return ok(
{"deleted": file_id, "storage_cleanup_pending": cleanup_pending},
"source file removed",
@@ -1964,6 +1966,7 @@ def stop(
) -> dict[str, Any]:
with api_errors():
store.stop_task(task_id)
+ biz_logger.info("data-process:stop success", taskId=task_id)
return ok(store.progress(task_id), "data process task stopped")
@@ -2005,7 +2008,9 @@ def confirm_results(
store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
with api_errors():
- return ok(store.confirm_results(task_id), "data process results confirmed")
+ result = store.confirm_results(task_id)
+ biz_logger.info("data-process:confirm-results success", taskId=task_id)
+ return ok(result, "data process results confirmed")
@router.put("/{task_id}/results/{result_id}")
@@ -2499,5 +2504,6 @@ def publish(
) -> dict[str, Any]:
with api_errors():
result = store.publish(task_id, payload.model_dump(mode="json"))
+ biz_logger.info("data-process:publish success", taskId=task_id, datasetId=result.get("dataset_id", ""))
message = "dataset published" if result["created"] else "dataset already published"
return ok(result, message)
diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py
index eb3ab1d..0bd84f5 100644
--- a/backend/app/api/v1/endpoints/platform.py
+++ b/backend/app/api/v1/endpoints/platform.py
@@ -471,6 +471,7 @@ async def _fine_tune_preflight_with_job_payload(
@router.post("/login")
+@op_log(module=OpModule.SYSTEM, action=OpAction.LOGIN, target_type="user", target_name_param="username")
async def login(payload: dict[str, Any] = Body(...), request: Request = None) -> dict[str, Any]:
store = get_platform_store()
ip = request.client.host if request and request.client else "unknown"
@@ -488,6 +489,7 @@ async def login(payload: dict[str, Any] = Body(...), request: Request = None) ->
@router.post("/logout")
+@op_log(module=OpModule.SYSTEM, action=OpAction.LOGOUT, target_type="user")
async def logout(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
store = get_platform_store()
session_id = payload.get("session_id", "")
@@ -717,11 +719,13 @@ async def users() -> dict[str, Any]:
@router.post("/users")
+@op_log(module=OpModule.SYSTEM, action=OpAction.CREATE, target_type="user", target_name_param="username")
async def create_user(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok(get_platform_store().create_user(payload))
@router.put("/users/{user_id}")
+@op_log(module=OpModule.SYSTEM, action=OpAction.UPDATE, target_type="user", target_name_param="user_id")
async def update_user(user_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
return ok(get_platform_store().update_user(user_id, payload))
@@ -730,6 +734,7 @@ async def update_user(user_id: str, payload: dict[str, Any] = Body(...)) -> dict
@router.delete("/users/{user_id}")
+@op_log(module=OpModule.SYSTEM, action=OpAction.DELETE, target_type="user", target_name_param="user_id")
async def delete_user(user_id: str, current_username: str | None = Query(default=None)) -> dict[str, Any]:
try:
get_platform_store().delete_user(user_id)
@@ -741,6 +746,7 @@ async def delete_user(user_id: str, current_username: str | None = Query(default
@router.post("/users/{user_id}/reset-password")
+@op_log(module=OpModule.SYSTEM, action=OpAction.UPDATE, target_type="user_password", target_name_param="user_id")
async def reset_user_password(
user_id: str,
payload: dict[str, Any] = Body(default={}),
@@ -756,6 +762,7 @@ async def reset_user_password(
@router.post("/users/me/password")
+@op_log(module=OpModule.SYSTEM, action=OpAction.UPDATE, target_type="user_password")
async def change_my_password(
payload: dict[str, Any] = Body(...),
current_user: dict = Depends(get_current_user),
@@ -819,6 +826,7 @@ async def trained_models(current_user: dict = Depends(get_current_user)) -> dict
@router.delete("/model-manage/trained-models/{model_id}")
+@op_log(module=OpModule.MODEL_MANAGE, action=OpAction.DELETE, target_type="trained_model", target_name_param="model_id")
async def delete_trained_model(model_id: str, type: str = Query(default="merged"), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not has_resource_access("trained_model", model_id, current_user, "delete"):
raise fail(403, "no permission to delete this trained model")
@@ -922,6 +930,7 @@ async def test_online_model(payload: dict[str, Any] = Body(...)) -> dict[str, An
@router.post("/model-manage")
+@op_log(module=OpModule.MODEL_MANAGE, action=OpAction.CREATE, target_type="model", target_name_param="name")
@audit_log(
action=AuditActions.CREATE_MODEL,
target_type="model",
@@ -950,6 +959,7 @@ async def model_detail(model_id: str, current_user: dict = Depends(get_current_u
@router.put("/model-manage/{model_id}")
+@op_log(module=OpModule.MODEL_MANAGE, action=OpAction.UPDATE, target_type="model", target_name_param="model_id")
@audit_log(
action=AuditActions.UPDATE_MODEL,
target_type="model",
@@ -966,6 +976,7 @@ async def update_model(model_id: str, payload: dict[str, Any] = Body(...), curre
@router.put("/model-manage/{model_id}/purpose")
+@op_log(module=OpModule.MODEL_MANAGE, action=OpAction.UPDATE, target_type="model_purpose", target_name_param="model_id")
async def update_model_purpose(model_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
# 基座模型(配置模型)只有管理员可以修改用途
if not is_admin(current_user):
@@ -977,6 +988,7 @@ async def update_model_purpose(model_id: str, payload: dict[str, Any] = Body(...
@router.delete("/model-manage/{model_id}")
+@op_log(module=OpModule.MODEL_MANAGE, action=OpAction.DELETE, target_type="model", target_name_param="model_id")
async def delete_model(model_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
# 基座模型(配置模型)只有管理员可以删除
if not is_admin(current_user):
@@ -1100,6 +1112,7 @@ async def dataset_version_content(file_id: str, version_id: str) -> dict[str, An
@router.post("/dataset-manage/versions/{file_id}")
+@op_log(module=OpModule.DATASET, action=OpAction.CREATE, target_type="dataset_version", target_name_param="file_id")
async def create_dataset_version(file_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
return ok(get_platform_store().create_file_version(file_id, payload))
@@ -1116,6 +1129,7 @@ async def activate_dataset_version(file_id: str, payload: dict[str, Any] = Body(
@router.delete("/dataset-manage/versions/{file_id}/{version_id}")
+@op_log(module=OpModule.DATASET, action=OpAction.DELETE, target_type="dataset_version", target_name_param="version_id")
async def delete_dataset_version(file_id: str, version_id: str) -> dict[str, Any]:
try:
return ok(get_platform_store().delete_file_version(file_id, version_id))
@@ -1281,6 +1295,7 @@ async def _sync_training_dataset_to_compute_node(
@router.post("/dataset-manage/upload/{dataset_id}")
+@op_log(module=OpModule.DATASET, action=OpAction.UPLOAD, target_type="dataset_file", target_name_param="dataset_id")
async def upload_dataset_files(
dataset_id: str,
files: list[UploadFile] = File(default=[]),
@@ -1403,6 +1418,7 @@ async def dataset_list(current_user: dict = Depends(get_current_user)) -> dict[s
@router.post("/dataset-manage")
+@op_log(module=OpModule.DATASET, action=OpAction.CREATE, target_type="dataset", target_name_param="name")
@audit_log(
action=AuditActions.CREATE_DATASET,
target_type="dataset",
@@ -1430,6 +1446,7 @@ async def dataset_detail(dataset_id: str, current_user: dict = Depends(get_curre
@router.put("/dataset-manage/{dataset_id}")
+@op_log(module=OpModule.DATASET, action=OpAction.UPDATE, target_type="dataset", target_name_param="dataset_id")
@audit_log(
action=AuditActions.UPDATE_DATASET,
target_type="dataset",
@@ -1491,6 +1508,7 @@ async def fine_tune_list(current_user: dict = Depends(get_current_user)) -> dict
@router.post("/fine-tune")
+@op_log(module=OpModule.FINE_TUNE, action=OpAction.CREATE, target_type="fine_tune", target_name_param="name")
@audit_log(
action=AuditActions.CREATE_FINE_TUNE,
target_type="fine_tune",
@@ -1662,6 +1680,7 @@ async def fine_tune_diagnostics(task_id: str) -> dict[str, Any]:
@router.put("/fine-tune/{task_id}")
+@op_log(module=OpModule.FINE_TUNE, action=OpAction.UPDATE, target_type="fine_tune", target_name_param="task_id")
async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not has_resource_access("fine-tune", task_id, current_user, "write"):
raise fail(403, "no permission to update this task")
@@ -1696,6 +1715,7 @@ async def stop_fine_tune_alt(task_id: str, current_user: dict = Depends(get_curr
@router.post("/fine-tune/{task_id}/retry")
+@op_log(module=OpModule.FINE_TUNE, action=OpAction.RETRY, target_type="fine_tune", target_name_param="task_id")
async def retry_fine_tune(task_id: str, payload: dict[str, Any] | None = Body(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
payload = payload or {}
@@ -1964,6 +1984,7 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
@router.delete("/model-eval/{task_id}")
+@op_log(module=OpModule.MODEL_EVAL, action=OpAction.DELETE, target_type="eval_task", target_name_param="task_id")
async def model_eval_delete(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not has_resource_access("eval", task_id, current_user, "delete"):
raise fail(403, "no permission to delete this eval task")
@@ -1980,6 +2001,7 @@ async def dimension_list() -> dict[str, Any]:
@router.post("/dimension")
+@op_log(module=OpModule.MODEL_EVAL, action=OpAction.CREATE, target_type="dimension", target_name_param="name")
async def dimension_create(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok(get_platform_store().create_dimension(payload))
@@ -1993,6 +2015,7 @@ async def dimension_detail(dimension_id: str) -> dict[str, Any]:
@router.put("/dimension/{dimension_id}")
+@op_log(module=OpModule.MODEL_EVAL, action=OpAction.UPDATE, target_type="dimension", target_name_param="dimension_id")
async def dimension_update(dimension_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
return ok(get_platform_store().update_dimension(dimension_id, payload))
@@ -2001,6 +2024,7 @@ async def dimension_update(dimension_id: str, payload: dict[str, Any] = Body(...
@router.delete("/dimension/{dimension_id}")
+@op_log(module=OpModule.MODEL_EVAL, action=OpAction.DELETE, target_type="dimension", target_name_param="dimension_id")
async def dimension_delete(dimension_id: str) -> dict[str, Any]:
get_platform_store().delete_dimension(dimension_id)
return ok({"deleted": dimension_id})
@@ -2019,6 +2043,7 @@ async def model_compare_list(current_user: dict = Depends(get_current_user)) ->
@router.post("/model-compare")
+@op_log(module=OpModule.INFERENCE, action=OpAction.CREATE, target_type="compare_task", target_name_param="name")
async def model_compare_create(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
payload.setdefault("created_by", current_user.get("id"))
model_ids = payload.get("model_ids") or payload.get("models") or []
@@ -2574,6 +2599,7 @@ async def compute_node_detail(node_id: str, current_user: dict = Depends(get_cur
@router.post("/compute/nodes")
+@op_log(module=OpModule.COMPUTE, action=OpAction.CREATE, target_type="compute_node", target_name_param="name")
async def create_compute_node(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
@@ -2586,6 +2612,7 @@ async def create_compute_node(payload: dict[str, Any] = Body(...), current_user:
@router.put("/compute/nodes/{node_id}")
+@op_log(module=OpModule.COMPUTE, action=OpAction.UPDATE, target_type="compute_node", target_name_param="node_id")
async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
@@ -2598,6 +2625,7 @@ async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...),
@router.delete("/compute/nodes/{node_id}")
+@op_log(module=OpModule.COMPUTE, action=OpAction.DELETE, target_type="compute_node", target_name_param="node_id")
async def delete_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
@@ -2904,6 +2932,7 @@ async def compute_job_detail(job_id: str) -> dict[str, Any]:
@router.post("/compute/jobs/{job_id}/stop")
+@op_log(module=OpModule.COMPUTE, action=OpAction.STOP, target_type="compute_job", target_name_param="job_id")
async def compute_job_stop(job_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
task = _task_for_compute_job(job_id)
if task and not has_resource_access("fine-tune", task["id"], current_user, "write"):
@@ -2942,6 +2971,7 @@ async def compute_job_logs(
@router.post("/compute/jobs/{job_id}/retry")
+@op_log(module=OpModule.COMPUTE, action=OpAction.RETRY, target_type="compute_job", target_name_param="job_id")
async def compute_job_retry(job_id: str, payload: dict[str, Any] | None = Body(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
payload = payload or {}
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index dd68d01..8d67558 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -50,7 +50,7 @@ def docs_kwargs(enabled: bool) -> dict[str, Any]:
@dataclass(frozen=True)
class Settings:
- app_name: str = os.getenv("APP_NAME", "YG Fine-Tune Platform API")
+ app_name: str = os.getenv("APP_NAME", "YG Zhilian API")
app_env: str = os.getenv("APP_ENV", "local")
route_prefix: str = os.getenv("MODELTF_ROUTE_PREFIX", "/modelTF")
app_mode: str = os.getenv("APP_MODE", "local")
diff --git a/backend/app/core/logging.py b/backend/app/core/logging.py
index 51453f3..5880834 100644
--- a/backend/app/core/logging.py
+++ b/backend/app/core/logging.py
@@ -1,162 +1,255 @@
from __future__ import annotations
-from contextvars import ContextVar
-from datetime import date, datetime, timedelta
import json
import logging
+import re
+import socket
import sys
+import time
+from contextvars import ContextVar
+from datetime import date, datetime, timedelta
from logging import Handler, LogRecord
from pathlib import Path
-import re
-import time
-from typing import Any, Callable, Optional
+from typing import Any, Callable
from uuid import uuid4
from fastapi import FastAPI, Request
from app.core.config import Settings, get_settings
+# ==================== 链路追踪 ContextVar ====================
+
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
+user_id_var: ContextVar[str] = ContextVar("user_id", default="")
+client_ip_var: ContextVar[str] = ContextVar("client_ip", default="")
-# ==================== 敏感数据脱敏规则 ====================
+# ==================== 敏感数据脱敏 ====================
-SENSITIVE_PATTERNS: dict[str, Callable | str] = {
- "token": "***",
- "password": "***",
- "access_token": "***",
- "refresh_token": "***",
- "secret_key": "***",
- "authorization": "***",
- "bearer": "***",
- "api_key": "***",
- "private_key": "***",
+SENSITIVE_KEYS: set[str] = {
+ "password", "token", "access_token", "refresh_token",
+ "secret_key", "authorization", "bearer", "api_key",
+ "private_key", "secret", "cookie",
}
-def mask_value(key: str, value: Any) -> str:
- """对单个值进行脱敏处理"""
+FULL_MASK_KEYS: set[str] = {
+ "password", "token", "access_token", "refresh_token",
+ "secret_key", "authorization", "bearer", "api_key",
+ "private_key", "secret", "cookie",
+}
+
+
+def _mask_phone(value: str) -> str:
+ """手机号脱敏:138****5678"""
+ if len(value) >= 11:
+ return value[:3] + "****" + value[-4:]
+ return value
+
+
+def _mask_id_card(value: str) -> str:
+ """身份证号脱敏:110***********1234"""
+ if len(value) >= 18:
+ return value[:3] + "***********" + value[-4:]
+ return value
+
+
+def mask_value(key: str, value: Any) -> Any:
+ """对单个值进行脱敏处理。"""
if value is None:
return ""
+ key_lower = key.lower()
+ if key_lower in FULL_MASK_KEYS:
+ return "***"
str_val = str(value)
-
- handler = SENSITIVE_PATTERNS.get(key)
- if callable(handler):
- return handler(str_val)
- elif isinstance(handler, str):
- # 支持正则替换模式,如 r"1\d{3}\d{4}"
- try:
- return re.sub(handler, "***", str_val)
- except re.error:
- return "***"
- return handler
+ # 手机号模式(11位数字,1开头)
+ if re.match(r"^1[3-9]\d{9}$", str_val):
+ return _mask_phone(str_val)
+ # 身份证模式(18位)
+ if re.match(r"^\d{17}[\dXx]$", str_val):
+ return _mask_id_card(str_val)
+ return value
def mask_sensitive_dict(data: dict) -> dict:
- """递归脱敏字典中的敏感字段"""
+ """递归脱敏字典中的敏感字段。"""
if not data or not isinstance(data, dict):
return data
-
- result = {}
+ result: dict[str, Any] = {}
for key, value in data.items():
- result[key] = mask_value(key, value)
+ if isinstance(value, dict):
+ result[key] = mask_sensitive_dict(value)
+ elif isinstance(value, list):
+ result[key] = [
+ mask_sensitive_dict(item) if isinstance(item, dict) else item
+ for item in value
+ ]
+ else:
+ result[key] = mask_value(key, value)
return result
def mask_sensitive_string(text: str) -> str:
- """从文本中脱敏常见敏感信息"""
+ """从文本中脱敏常见敏感信息。"""
if not text:
return text
-
- patterns = [
- (r'Bearer\s+[A-Za-z0-9\-._]+', '***'),
- (r'token\s*[:=]\s*', '***'),
- (r'password\s*[:=]\s*', '***'),
- (r'secret[_-]?key\s*[:=]', '***'),
- (r'api[-_]?key\s*[:=]', '***'),
- (r'private[_-]?key\s*[:=]', '***'),
- (r'\d{11}', r'\d{3}\*\d{4}'), # 手机号/身份证
- (r'1[3-9]\d{9}', r'1\*{3}\*{4}'), # 手机号
+ patterns: list[tuple[str, str]] = [
+ (r"Bearer\s+[A-Za-z0-9\-._]+", "Bearer ***"),
+ (r"(?i)token\s*[:=]\s*\S+", "token=***"),
+ (r"(?i)password\s*[:=]\s*\S+", "password=***"),
+ (r"(?i)secret[_-]?key\s*[:=]\s*\S+", "secret_key=***"),
+ (r"(?i)api[-_]?key\s*[:=]\s*\S+", "api_key=***"),
+ (r"(?i)private[_-]?key\s*[:=]\s*\S+", "private_key=***"),
+ (r"(?i)authorization\s*[:=]\s*\S+", "authorization=***"),
]
-
for pattern, replacement in patterns:
- try:
- text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
- except re.error:
- pass
+ text = re.sub(pattern, replacement, text)
+ # 手机号脱敏
+ text = re.sub(r"\b1[3-9]\d{9}\b", lambda m: _mask_phone(m.group()), text)
return text
-# ==================== RequestId Filter ====================
+# ==================== 大对象截断 ====================
+
+MAX_FIELD_SIZE = 1024 # 超过 1KB 的内容自动截断
+
+
+def truncate_large_value(value: Any, max_size: int = MAX_FIELD_SIZE) -> Any:
+ """超过 max_size 的字符串自动截断(前 500 + 后 500)。"""
+ if isinstance(value, str) and len(value) > max_size:
+ half = max_size // 2
+ return value[:half] + f"...[truncated {len(value) - max_size} chars]..." + value[-half:]
+ if isinstance(value, dict):
+ return {k: truncate_large_value(v, max_size) for k, v in value.items()}
+ if isinstance(value, list):
+ return [truncate_large_value(v, max_size) for v in value]
+ return value
+
+
+# ==================== TraceId Filter ====================
+
+class TraceIdFilter(logging.Filter):
+ """自动注入 traceId / userId / clientIp 到每条日志记录。"""
-class RequestIdFilter(logging.Filter):
def filter(self, record: LogRecord) -> bool:
- record.request_id = request_id_var.get()
+ record.traceId = request_id_var.get()
+ record.userId = user_id_var.get("")
+ record.clientIp = client_ip_var.get("")
+ record.host = getattr(self, "_host", None) or socket.gethostname()
+ record.app = getattr(self, "_app", "yg-ft-platform")
+ record.env = getattr(self, "_env", "dev")
return True
+ def set_context(self, app: str, env: str, host: str) -> None:
+ self._app = app
+ self._env = env
+ self._host = host
-# ==================== Enhanced JSON Formatter ====================
+
+# ==================== JSON Formatter ====================
class JsonLogFormatter(logging.Formatter):
"""
- 增强的 JSON 日志格式化器,支持结构化字段输出。
-
+ 生产级 JSON 日志格式化器,符合方案文档 §3.2 字段规范。
+
输出示例:
{
- "@timestamp": "2026-08-17T18:30:00.123Z",
+ "@timestamp": "2026-08-19T10:30:45.123+08:00",
"level": "INFO",
- "logger": "dataset.router",
+ "logger": "app.api.v1.endpoints.platform",
+ "traceId": "abc-123-def-456",
+ "userId": "u_admin",
"message": "数据集创建成功",
- "module": "dataset.router",
- "function": "create_dataset",
- "file": "dataset/router.py",
- "line": 45,
- "process": 12345,
+ "fields": {"datasetId": "ds_001", "costMs": 23},
+ "file": "platform.py:156",
"thread": "MainThread",
- "request_id": "req-abc123",
- "user_id": "u_admin",
- "client_ip": "192.168.1.100",
- "extra": {...}
+ "host": "pod-7x9k2",
+ "app": "yg-ft-platform",
+ "env": "dev"
}
"""
+ # 标准 LogRecord 属性名集合,用于区分 extra 字段
+ _STD_ATTRS: set[str] = set(vars(logging.LogRecord("", 0, "", 0, "", None, None)).keys()) | {
+ "traceId", "userId", "clientIp", "host", "app", "env",
+ "request_id", "user_id", "client_ip",
+ "asctime", "message", "module", "function", "process",
+ "thread", "threadName", "levelname", "levelno", "name",
+ "pathname", "filename", "lineno", "funcName", "created",
+ "msecs", "relativeCreated", "exc_info", "exc_text",
+ "stack_info", "msg", "args", "processName", "process",
+ }
+
def format(self, record: LogRecord) -> str:
+ # 时间戳:ISO8601 带时区
+ timestamp = datetime.fromtimestamp(record.created).astimezone().isoformat(
+ timespec="milliseconds"
+ )
+
payload: dict[str, Any] = {
- "@timestamp": datetime.fromtimestamp(record.created).astimezone().isoformat(
- timespec="milliseconds"
- ),
+ "@timestamp": timestamp,
"level": record.levelname,
"logger": record.name,
+ "traceId": getattr(record, "traceId", "-"),
"message": record.getMessage(),
- "module": record.module,
- "function": record.funcName,
- "file": record.pathname,
- "line": record.lineno,
- "process": record.process,
- "thread": record.thread,
- "thread_name": record.threadName,
- "request_id": getattr(record, "request_id", "-"),
+ "file": f"{Path(record.pathname).name}:{record.lineno}",
+ "thread": record.threadName,
+ "host": getattr(record, "host", ""),
+ "app": getattr(record, "app", ""),
+ "env": getattr(record, "env", ""),
}
-
- # 从 record 中提取额外字段(通过 extra 参数传入)
- for attr in ("user_id", "client_ip", "target_type", "target_id",
- "duration_ms", "status_code", "error"):
+
+ # userId(业务必填,未登录可为空)
+ user_id = getattr(record, "userId", "") or getattr(record, "user_id", "")
+ if user_id:
+ payload["userId"] = user_id
+
+ # clientIp
+ client_ip = getattr(record, "clientIp", "") or getattr(record, "client_ip", "")
+ if client_ip:
+ payload["clientIp"] = client_ip
+
+ # 提取结构化业务字段:只收集通过 extra 传入的非标准属性
+ fields: dict[str, Any] = {}
+ for attr in dir(record):
+ if attr.startswith("_"):
+ continue
+ if attr in self._STD_ATTRS:
+ continue
+ if attr in ("traceId", "userId", "clientIp", "host", "app", "env"):
+ continue
val = getattr(record, attr, None)
- if val is not None:
- payload[attr] = val
-
- # 处理异常信息
- if record.exc_info:
+ if val is not None and not callable(val):
+ fields[attr] = truncate_large_value(val)
+ if fields:
+ payload["fields"] = mask_sensitive_dict(fields)
+
+ # ERROR 级别额外字段
+ if record.levelname == "ERROR" or record.exc_info:
+ error_obj: dict[str, Any] = {
+ "type": type(record.exc_info[1]).__name__ if record.exc_info and record.exc_info[1] else "Error",
+ "message": record.getMessage(),
+ }
+ if record.exc_info:
+ error_obj["stack_trace"] = self.formatException(record.exc_info)
+ if record.stack_info:
+ error_obj["stack_trace"] = self.formatStack(record.stack_info)
+ payload["error"] = error_obj
+
+ # 兼容旧字段名 exception
+ if record.exc_info and "error" not in payload:
payload["exception"] = self.formatException(record.exc_info)
- if record.stack_info:
- payload["stack"] = self.formatStack(record.stack_info)
-
+
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
# ==================== DateSizeRotatingFileHandler ====================
-# (保持不变,已有实现)
class DateSizeRotatingFileHandler(Handler):
- """Rotate log files by date and size while keeping date in every file name."""
+ """按日期+大小滚动的文件日志处理器。
+
+ - 按天创建文件,文件名包含日期
+ - 单文件超过 max_bytes 时自动滚动(带序号后缀)
+ - 自动清理超过 retention_days 的旧日志
+ """
def __init__(
self,
@@ -214,10 +307,8 @@ class DateSizeRotatingFileHandler(Handler):
today = date.today()
if not force and self._stream and self._current_date == today:
return
-
if self._stream and not self._stream.closed:
self._stream.close()
-
self._current_date = today
self._current_path = self._dated_path(today)
self._stream = self._current_path.open("a", encoding=self.encoding)
@@ -232,11 +323,9 @@ class DateSizeRotatingFileHandler(Handler):
def _rotate_by_size(self) -> None:
if not self._current_path or not self._current_path.exists():
return
-
if self._stream and not self._stream.closed:
self._stream.close()
self._stream = None
-
stem = self._current_path.stem
suffix = self._current_path.suffix
index = 1
@@ -250,7 +339,6 @@ class DateSizeRotatingFileHandler(Handler):
def _cleanup_expired_files(self) -> None:
if self.retention_days <= 0:
return
-
cutoff = date.today() - timedelta(days=self.retention_days - 1)
pattern = re.compile(
rf"^{re.escape(self.file_prefix)}-(\d{{4}}-\d{{2}}-\d{{2}})(?:\.\d+)?\.log$"
@@ -264,98 +352,125 @@ class DateSizeRotatingFileHandler(Handler):
path.unlink(missing_ok=True)
-# ==================== Structured Logger 封装 ====================
+# ==================== StructuredLogger 封装 ====================
class StructuredLogger:
"""
- 结构化日志记录器,提供统一的日志接口。
-
+ 结构化日志记录器,提供符合方案文档 §4.2 的 5W1H 日志接口。
+
使用方式:
- logger = get_structured_logger('dataset.router')
- logger.info('创建数据集', dataset_id='ds_123')
+ logger = get_structured_logger('app.api.dataset')
+ logger.info('数据集创建成功', datasetId='ds_001', costMs=23)
"""
-
+
def __init__(self, name: str, module: str = ""):
self.logger = logging.getLogger(name)
self.name = name
self.module = module
-
+
@property
def trace_id(self) -> str:
return request_id_var.get("-")
-
- def info(self, message: str, **extra: Any) -> None:
- self._log("INFO", message, **extra)
- def warning(self, message: str, **extra: Any) -> None:
- self._log("WARNING", message, **extra)
+ def info(self, message: str, **fields: Any) -> None:
+ self._log(logging.INFO, message, **fields)
- def error(self, message: str, **extra: Any) -> None:
- self._log("ERROR", message, **extra)
+ def warning(self, message: str, **fields: Any) -> None:
+ self._log(logging.WARNING, message, **fields)
- def debug(self, message: str, **extra: Any) -> None:
- self._log("DEBUG", message, **extra)
+ def error(self, message: str, **fields: Any) -> None:
+ self._log(logging.ERROR, message, **fields)
- def _log(self, level: str, message: str, **extra: Any) -> None:
- """统一日志记录方法"""
- log_entry: dict[str, Any] = {
- "timestamp": datetime.utcnow().isoformat(),
- "level": level,
- "logger": self.name,
- "module": self.module,
- "message": message,
- "trace_id": self.trace_id,
- "extra": extra,
- }
- self.logger.log(getattr(logging, level, logging.INFO), json.dumps(log_entry, ensure_ascii=False, default=str))
+ def debug(self, message: str, **fields: Any) -> None:
+ self._log(logging.DEBUG, message, **fields)
+
+ def _log(self, level: int, message: str, **fields: Any) -> None:
+ """统一日志记录方法,通过 extra 传递结构化字段。"""
+ extra: dict[str, Any] = {}
+ if self.module:
+ extra["module"] = self.module
+ # 脱敏 + 截断
+ for k, v in fields.items():
+ extra[k] = truncate_large_value(v)
+ self.logger.log(level, message, extra=extra, stack_info=False)
def get_structured_logger(name: str, module: str = "") -> StructuredLogger:
- """获取结构化日志记录器"""
+ """获取结构化日志记录器。"""
return StructuredLogger(name, module)
# ==================== 快捷函数 ====================
def get_logger(name: str) -> logging.Logger:
- """获取标准 Python logger"""
+ """获取标准 Python logger。"""
return logging.getLogger(name)
def set_request_id(request_id: str) -> None:
- """设置当前请求的追踪 ID"""
+ """设置当前请求的追踪 ID。"""
request_id_var.set(request_id)
+def set_user_context(user_id: str = "", client_ip: str = "") -> None:
+ """设置当前请求的用户上下文(在鉴权后调用)。"""
+ if user_id:
+ user_id_var.set(user_id)
+ if client_ip:
+ client_ip_var.set(client_ip)
+
+
+# ==================== 请求日志中间件 ====================
+
def setup_request_logging(app: FastAPI) -> None:
- """配置 FastAPI 请求日志中间件"""
+ """配置 FastAPI 请求日志中间件,符合方案文档 §五(链路追踪)和 §十(访问日志)。"""
logger = get_logger("app.access")
@app.middleware("http")
async def request_logging_middleware(request: Request, call_next): # type: ignore[no-untyped-def]
- request_id = request.headers.get("X-Request-ID") or str(uuid4())
- token = request_id_var.set(request_id)
+ # 入口生成 traceId(优先使用前端传入的 X-Trace-Id)
+ trace_id = request.headers.get("X-Trace-Id") or request.headers.get("X-Request-ID") or str(uuid4())
+ token = request_id_var.set(trace_id)
started_at = time.perf_counter()
+
+ # 提取客户端 IP
+ client_ip = "-"
+ if request.client:
+ client_ip = request.client.host
+ # 支持反向代理传递的真实 IP
+ forwarded_for = request.headers.get("X-Forwarded-For", "")
+ if forwarded_for:
+ client_ip = forwarded_for.split(",")[0].strip()
+ client_ip_var.set(client_ip)
+
try:
response = await call_next(request)
elapsed_ms = (time.perf_counter() - started_at) * 1000
+ # 噪声路径降级为 DEBUG(健康检查等)
noisy_paths = ("/health", "/system-info", "/compute/jobs/", "/model-eval/", "/model-compare/")
- log_method = logger.debug if request.method == "GET" and response.status_code < 400 else logger.info
- if any(request.url.path.endswith(path) or path in request.url.path for path in noisy_paths) and response.status_code < 400:
+ log_method = logger.info
+ if any(request.url.path.endswith(p) or p in request.url.path for p in noisy_paths) and response.status_code < 400:
log_method = logger.debug
- if response.status_code >= 400:
+ if response.status_code >= 500:
+ log_method = logger.error
+ elif response.status_code >= 400:
log_method = logger.warning
+
+ # 结构化访问日志
log_method(
- "request completed method=%s path=%s status_code=%s duration_ms=%.2f client=%s",
- request.method,
- request.url.path,
- response.status_code,
- elapsed_ms,
- request.client.host if request.client else "-",
+ "request completed",
+ extra={
+ "request_method": request.method,
+ "request_path": request.url.path,
+ "status_code": response.status_code,
+ "duration_ms": round(elapsed_ms, 2),
+ "client_ip": client_ip,
+ "user_agent": request.headers.get("User-Agent", "")[:200],
+ },
)
- # 5xx 系统错误自动写入操作日志(未被 @op_log 覆盖的系统级异常)
+ # 5xx 系统错误自动写入操作日志
if response.status_code >= 500:
try:
from app.core.op_log import log_operation, OpModule, OpStatus
@@ -373,18 +488,23 @@ def setup_request_logging(app: FastAPI) -> None:
duration_ms=elapsed_ms,
)
except Exception:
- pass # 日志写入失败不影响主流程
+ pass
- response.headers["X-Request-ID"] = request_id
+ response.headers["X-Trace-Id"] = trace_id
+ response.headers["X-Request-ID"] = trace_id
return response
+
except Exception:
elapsed_ms = (time.perf_counter() - started_at) * 1000
- logger.exception(
- "request failed method=%s path=%s duration_ms=%.2f client=%s",
- request.method,
- request.url.path,
- elapsed_ms,
- request.client.host if request.client else "-",
+ logger.error(
+ "request failed",
+ extra={
+ "request_method": request.method,
+ "request_path": request.url.path,
+ "duration_ms": round(elapsed_ms, 2),
+ "client_ip": client_ip,
+ },
+ exc_info=True,
)
# 未被捕获的异常,写入操作日志
@@ -406,7 +526,7 @@ def setup_request_logging(app: FastAPI) -> None:
duration_ms=elapsed_ms,
)
except Exception:
- pass # 日志写入失败不影响主流程
+ pass
raise
finally:
@@ -416,53 +536,122 @@ def setup_request_logging(app: FastAPI) -> None:
# ==================== 配置函数 ====================
def configure_logging(settings: Settings | None = None) -> None:
+ """
+ 生产级日志配置,符合方案文档 §二(分类分流)和 §六(性能安全)。
+
+ 日志分类:
+ - 业务日志 (app-biz): INFO+ 业务流程(保留 7 天)
+ - 系统日志 (app-sys): 框架/中间件日志(保留 7 天)
+ - 访问日志 (app-access): HTTP 请求日志(保留 15 天)
+ - 错误日志 (app-error): ERROR 级别(保留 30 天)
+ """
settings = settings or get_settings()
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(settings.log_level.upper())
+ # ---- Formatter ----
console_formatter = logging.Formatter(
fmt=(
"%(asctime)s | %(levelname)s | pid=%(process)d | %(threadName)s | "
- "request_id=%(request_id)s | %(name)s | %(pathname)s:%(lineno)d | %(message)s"
+ "traceId=%(traceId)s | %(name)s | %(pathname)s:%(lineno)d | %(message)s"
),
datefmt="%Y-%m-%d %H:%M:%S",
)
json_formatter = JsonLogFormatter()
- request_filter = RequestIdFilter()
+ # ---- TraceIdFilter(全局注入 traceId/userId/host/app/env)----
+ trace_filter = TraceIdFilter()
+ trace_filter.set_context(
+ app=settings.app_name,
+ env=settings.app_env,
+ host=socket.gethostname(),
+ )
+
+ # ---- 控制台 Handler ----
console_handler = logging.StreamHandler()
console_handler.setFormatter(console_formatter)
- console_handler.addFilter(request_filter)
+ console_handler.addFilter(trace_filter)
- file_handler = DateSizeRotatingFileHandler(
+ # ---- 业务日志文件 Handler (app-biz) ----
+ biz_file_handler = DateSizeRotatingFileHandler(
log_dir=settings.log_dir,
- file_prefix=settings.log_file_prefix,
+ file_prefix="app-biz",
max_bytes=settings.log_max_bytes,
- retention_days=settings.log_retention_days,
+ retention_days=7,
)
- file_handler.setFormatter(json_formatter)
- file_handler.addFilter(request_filter)
+ biz_file_handler.setFormatter(json_formatter)
+ biz_file_handler.addFilter(trace_filter)
+ # ---- 访问日志文件 Handler (app-access) ----
+ access_file_handler = DateSizeRotatingFileHandler(
+ log_dir=settings.log_dir,
+ file_prefix="app-access",
+ max_bytes=settings.log_max_bytes,
+ retention_days=15,
+ )
+ access_file_handler.setFormatter(json_formatter)
+ access_file_handler.addFilter(trace_filter)
+
+ # ---- 错误日志文件 Handler (app-error) ----
error_file_handler = DateSizeRotatingFileHandler(
log_dir=settings.log_dir,
- file_prefix=settings.log_error_file_prefix,
+ file_prefix="app-error",
max_bytes=settings.log_max_bytes,
- retention_days=settings.log_retention_days,
+ retention_days=30,
)
error_file_handler.setLevel(logging.ERROR)
error_file_handler.setFormatter(json_formatter)
- error_file_handler.addFilter(request_filter)
+ error_file_handler.addFilter(trace_filter)
+ # ---- 注册 Handler ----
root_logger.addHandler(console_handler)
- root_logger.addHandler(file_handler)
+ root_logger.addHandler(biz_file_handler)
+ root_logger.addHandler(access_file_handler)
root_logger.addHandler(error_file_handler)
- for logger_name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
- logger = logging.getLogger(logger_name)
- logger.handlers.clear()
- logger.propagate = True
+ # ---- 访问日志 Logger 独立路由到访问日志文件 ----
+ access_logger = logging.getLogger("app.access")
+ access_logger.propagate = False # 不向 root 传播,避免重复写入业务日志
+ access_logger.addHandler(console_handler)
+ access_logger.addHandler(access_file_handler)
+ # 访问日志中的 ERROR 也要进错误日志
+ access_logger.addHandler(error_file_handler)
+ # ---- 框架类 Logger 降级 ----
+ for logger_name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
+ lg = logging.getLogger(logger_name)
+ lg.handlers.clear()
+ lg.propagate = True
+
+ # 框架类日志归入系统日志,生产环境设为 WARN
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
logging.getLogger("psycopg.pool").setLevel(logging.ERROR)
+ logging.getLogger("httpx").setLevel(logging.WARNING)
+
+ # ---- 兼容旧文件前缀(向后兼容)----
+ # 如果配置了旧的 log_file_prefix,也创建一个对应的 handler
+ if settings.log_file_prefix and settings.log_file_prefix != "app-biz":
+ legacy_file_handler = DateSizeRotatingFileHandler(
+ log_dir=settings.log_dir,
+ file_prefix=settings.log_file_prefix,
+ max_bytes=settings.log_max_bytes,
+ retention_days=settings.log_retention_days,
+ )
+ legacy_file_handler.setFormatter(json_formatter)
+ legacy_file_handler.addFilter(trace_filter)
+ root_logger.addHandler(legacy_file_handler)
+
+ # 旧错误日志前缀兼容
+ if settings.log_error_file_prefix and settings.log_error_file_prefix != "app-error":
+ legacy_error_handler = DateSizeRotatingFileHandler(
+ log_dir=settings.log_dir,
+ file_prefix=settings.log_error_file_prefix,
+ max_bytes=settings.log_max_bytes,
+ retention_days=settings.log_retention_days,
+ )
+ legacy_error_handler.setLevel(logging.ERROR)
+ legacy_error_handler.setFormatter(json_formatter)
+ legacy_error_handler.addFilter(trace_filter)
+ root_logger.addHandler(legacy_error_handler)
diff --git a/backend/app/core/op_log.py b/backend/app/core/op_log.py
index 2020356..3a4b780 100644
--- a/backend/app/core/op_log.py
+++ b/backend/app/core/op_log.py
@@ -31,10 +31,11 @@ from typing import Any, Callable, Optional, TypeVar
from fastapi import Request
-from app.core.logging import get_logger, request_id_var
+from app.core.logging import get_logger, get_structured_logger, request_id_var
from app.db.platform_store import get_platform_store, new_id, utcnow
logger = get_logger("app.op_log")
+biz_logger = get_structured_logger("app.biz")
F = TypeVar("F", bound=Callable[..., Any])
@@ -341,20 +342,44 @@ def _write_log(
trace_id: str,
duration_ms: float,
) -> None:
- """写入操作日志到数据库"""
+ """写入操作日志到数据库 + 文件日志"""
+ user_id = user.get("id") if user else None
+ username = user.get("username") if user else None
+ client_ip = None
+ req_method = None
+ req_path = None
+ if request:
+ client_ip = request.client.host if request.client else None
+ req_method = request.method
+ req_path = request.url.path
+
+ # ---- 写文件日志(app-biz)----
+ log_fields = {
+ "bizModule": module,
+ "action": action,
+ "targetType": target_type or "",
+ "targetId": target_id or "",
+ "targetName": target_name or "",
+ "opStatus": status,
+ "durationMs": round(duration_ms, 2),
+ "username": username or "",
+ }
+ if detail:
+ log_fields["detail"] = detail[:500]
+ if error_message:
+ log_fields["errorMessage"] = error_message[:500]
+ if error_type:
+ log_fields["errorType"] = error_type
+
+ if status == OpStatus.SUCCESS:
+ biz_logger.info(f"{module}:{action} {status} - {target_name or target_id or ''}", **log_fields)
+ elif status == OpStatus.FAILURE:
+ biz_logger.error(f"{module}:{action} {status} - {target_name or target_id or ''} - {error_type}: {error_message[:200]}", **log_fields)
+
+ # ---- 写数据库 ----
try:
store = get_platform_store()
log_id = new_id("op")
- user_id = user.get("id") if user else None
- username = user.get("username") if user else None
- client_ip = None
- req_method = None
- req_path = None
- if request:
- client_ip = request.client.host if request.client else None
- req_method = request.method
- req_path = request.url.path
-
with store.connect() as conn:
conn.execute(
"""
@@ -379,4 +404,4 @@ def _write_log(
),
)
except Exception:
- logger.error("写入操作日志失败 module=%s action=%s", module, action, exc_info=True)
+ logger.error("写入操作日志到数据库失败 module=%s action=%s", module, action, exc_info=True)
diff --git a/backend/app/db/sql/000_full_init.sql b/backend/app/db/sql/000_full_init.sql
index 222e7db..0aaf969 100644
--- a/backend/app/db/sql/000_full_init.sql
+++ b/backend/app/db/sql/000_full_init.sql
@@ -1,5 +1,5 @@
-- ============================================================================
--- YG Fine-Tune Platform — PostgreSQL 完整初始化脚本(一键建库建表)
+-- YG Zhilian — PostgreSQL 完整初始化脚本(一键建库建表)
-- ============================================================================
-- 用途:切换到新的 PG 数据集时,一次性创建平台运行所需的全部数据库对象与
-- 基础种子数据(幂等,可重复执行)。
diff --git a/compute/api/main.py b/compute/api/main.py
index 6d47c60..6b13470 100644
--- a/compute/api/main.py
+++ b/compute/api/main.py
@@ -23,7 +23,7 @@ from compute.engines.llama_factory.inference import get_inference_session
def create_app() -> FastAPI:
- app = FastAPI(title="YG Fine-Tune Compute API", **docs_kwargs())
+ app = FastAPI(title="YG Zhilian Compute API", **docs_kwargs())
jobs: dict[str, dict[str, Any]] = {}
cache_locks: dict[str, asyncio.Lock] = {}
route_prefix = os.getenv("MODELTF_ROUTE_PREFIX", "/modelTF").rstrip("/") or "/modelTF"
diff --git a/docker/app/docker-compose.yml b/docker/app/docker-compose.yml
index f57975b..59279d1 100644
--- a/docker/app/docker-compose.yml
+++ b/docker/app/docker-compose.yml
@@ -40,7 +40,7 @@ services:
- "${BACKEND_API_PORT:-17861}:8000"
environment:
APP_ENV: ${APP_ENV:-prod}
- APP_NAME: ${APP_NAME:-YG Fine-Tune Platform API}
+ APP_NAME: ${APP_NAME:-YG Zhilian API}
MODELTF_ROUTE_PREFIX: ${MODELTF_ROUTE_PREFIX:-/modelTF}
ENABLE_DOCS: ${ENABLE_DOCS:-false}
CORS_ALLOW_ORIGINS: ${CORS_ALLOW_ORIGINS:-http://localhost:16801,http://127.0.0.1:16801}
diff --git a/docs/governance-user-guide.md b/docs/governance-user-guide.md
index 618a39a..18afbf8 100644
--- a/docs/governance-user-guide.md
+++ b/docs/governance-user-guide.md
@@ -2,7 +2,7 @@
> 版本:v1.1
> 日期:2026-08-13
-> 适用版本:YG Fine-Tune Platform v1.0+
+> 适用版本:YG Zhilian v1.0+
> 更新说明:移除页面权限码设计,改为基于角色的简化权限模型
---
diff --git a/docs/security-hardening.md b/docs/security-hardening.md
index b15fdd2..bf8a35b 100644
--- a/docs/security-hardening.md
+++ b/docs/security-hardening.md
@@ -150,7 +150,7 @@ def docs_kwargs() -> dict[str, Any]:
return {"docs_url": None, "redoc_url": None, "openapi_url": None}
# main.py
-app = FastAPI(title="YG Fine-Tune Compute API", **docs_kwargs())
+app = FastAPI(title="YG Zhilian Compute API", **docs_kwargs())
```
**判定顺序(优先级从高到低)**:
diff --git a/docs/生产级日志系统方案.md b/docs/生产级日志系统方案.md
index 4eec059..4ab3e49 100644
--- a/docs/生产级日志系统方案.md
+++ b/docs/生产级日志系统方案.md
@@ -1,412 +1,489 @@
-# 生产级日志系统设计方案
-
-> 版本:v1.0
-> 日期:2026-08-17
-> 状态:待评审
+好的,这是一份可以直接放在项目根目录的 `日志规范要求.md`,涵盖**格式标准、分类分级、内容规范、链路追踪、性能安全、运维告警**六大模块,每条规范都配有正反例,你的团队照着这个写代码就行。
---
-## 一、现状分析
+# 生产级日志规范要求
-### 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、密码等可能明文记录 | 🔴 高 |
-| **无日志聚合查询** | 无法按用户/时间范围/操作类型快速检索 | 🟡 中 |
-| **无告警通知** | 系统异常无法主动推送通知 | 🟡 中 |
-| **日志文件无归档策略** | 只有简单的过期删除,无压缩归档 | 🟢 低 |
+> 版本:v2.0 | 适用于所有后端服务(Python/Java/Go/Node.js)
---
-## 二、设计目标
+## 一、核心原则
-### 2.1 核心原则
-
-1. **结构化** - 日志有固定 schema,便于机器解析和查询
-2. **可追溯** - 每个请求有唯一 ID,可串联完整调用链路
-3. **分级输出** - 不同环境输出不同级别,生产环境不输出 DEBUG
-4. **安全合规** - 敏感数据自动脱敏(token、密码、手机号等)
-5. **高性能** - 日志写入不影响业务接口性能(异步写入)
-6. **可观测** - 支持快速检索、统计、告警
-
-### 2.2 日志分级标准
-
-| 级别 | 使用场景 | 示例 | 生产环境 |
-|------|----------|------|:--------:|
-| **DEBUG** | 开发调试 | 变量值、SQL 语句、完整堆栈 | ❌ 不输出 |
-| **INFO** | 正常流程记录 | 任务创建成功、用户登录 | ✅ 记录 |
-| **WARNING** | 可恢复异常 | 重试操作、参数校验失败、资源不足 | ✅ 记录 |
-| **ERROR** | 需要人工介入 | 数据库连接失败、第三方 API 超时 | ✅ 记录 + 告警 |
-| **CRITICAL** | 系统不可用 | 磁盘满、主节点宕机 | ✅ 记录 + 立即告警 |
+| 原则 | 说明 |
+|------|------|
+| **结构化** | 所有日志必须输出为 JSON 格式,便于自动化采集和分析 |
+| **可追踪** | 每个请求链路必须有唯一的 `traceId`,贯穿全流程 |
+| **有上下文** | 每条日志必须包含足够的业务信息,能独立理解发生了什么 |
+| **高性能** | 异步打印,禁止在业务主流程中同步写磁盘 |
+| **安全合规** | 敏感信息自动脱敏,禁止打印密码、token、身份证号等 |
+| **可告警** | ERROR 日志必须触发实时告警,且有明确的错误分类 |
---
-## 三、技术方案
+## 二、日志分类
-### 3.1 整体架构
+生产环境必须按用途分流存储,**禁止所有日志混写在同一文件**:
-```
-┌─────────────────────────────────────────────────────────────────────┐
-│ 应用层 (Application Layer) │
-├─────────────────────────────────────────────────────────────────────┤
-│ │
-│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
-│ │ 数据集管理 │ │ 微调训练 │ │ 模型推理 │ │ 用户认证 │ ... │
-│ └─────┬────┘ └─────┬────┘ └─────┬────┘ └─────┬────┘ │
-│ │ │ │ │ │
-│ └────────────┴───────────┴──────────┘ │
-│ ▼ │
-│ ┌──────────────┐ │
-│ │ Structured │ ← 结构化日志中间件 │
-│ │ Logger │ │
-│ └──────┬───────┘ │
-│ │ │
-│ ┌────────────┬────────────┬─────────────┐ │
-│ ▼ ▼ ▼ │ │
-│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐ │
-│ │ Console │ │ File │ │ 审计DB │ │ 告警 │ │
-│ │ (开发) │ │ (JSON) │ │ (PG) │ │(可选) │ │
-│ └──────────┘ └──────────┘ └──────────┘ └────────┘ │
-│ │
-└─────────────────────────────────────────────────────────────┘
- │
- ▼
-┌─────────────────────────────────────────────────────────────┐
-│ 可观测层 (Observability) │
-├─────────────────────────────────────────────────────────────┤
-│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
-│ │ Grafana │ │ Kibana │ │ PagerDuty │ ... │
-│ │ (查询) │ │ (分析) │ │ (告警) │ │
-│ └───────────┘ └───────────┘ └───────────┘ │
-└─────────────────────────────────────────────────────────────┘
-```
+| 分类 | 文件名示例 | 用途 | 保留周期 |
+|------|-----------|------|----------|
+| **业务日志** | `app-biz.log` | 记录核心业务流程(订单、支付、登录、任务状态变更等) | 7天热存 + 30天冷存 |
+| **系统日志** | `app-sys.log` | 记录框架、中间件、连接池、GC、线程池状态 | 7天 |
+| **访问日志** | `app-access.log` | 记录所有 HTTP/RPC 请求的入参、出参、耗时 | 15天(用于审计) |
+| **错误日志** | `app-error.log` | **仅记录 ERROR 级别**,含完整堆栈 | 30天(用于复盘) |
-### 3.2 日志 Schema 设计
+**配置要点**:
+- 业务日志和错误日志必须独立文件,便于快速定位异常
+- 框架类日志(如 `httpx`、`urllib3`)归入系统日志,且生产环境设为 WARN 级别
-#### 3.2.1 应用日志 (app.log)
+---
+
+## 三、日志格式标准
+
+### 3.1 统一 JSON 格式
+
+所有日志必须输出为以下 JSON 结构,**字段名不得随意变更**:
```json
{
- "timestamp": "2026-08-17T10:30:00.000Z",
+ "@timestamp": "2026-08-19T10:30:45.123+08:00",
"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"
+ "logger": "com.order.service.OrderService",
+ "traceId": "abc-123-def-456",
+ "spanId": "span-001",
+ "userId": "U10086",
+ "message": "订单状态更新成功",
+ "fields": {
+ "orderId": "ORD-20260819-001",
+ "fromStatus": "PENDING",
+ "toStatus": "PAID",
+ "costMs": 23,
+ "retryCount": 0
},
- "module": "dataset.router",
- "function": "create_dataset",
- "message": "数据集创建成功",
- "extra": {
- "dataset_id": "ds_abc123",
- "dataset_name": "训练数据"
- },
- "duration_ms": 125,
- "status_code": 200,
- "error": null
+ "file": "OrderService.java:156",
+ "thread": "http-nio-8080-exec-8",
+ "host": "pod-order-7x9k2",
+ "app": "order-service",
+ "env": "prod"
}
```
-#### 3.2.2 审计日志 (audit_logs 表)
+### 3.2 字段说明
-```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 -- 扩展信息
-);
+| 字段 | 类型 | 必填 | 说明 |
+|------|------|------|------|
+| `@timestamp` | string | ✅ | ISO8601 格式,带时区(如 `+08:00`) |
+| `level` | string | ✅ | DEBUG / INFO / WARNING / ERROR |
+| `logger` | string | ✅ | 日志记录器名称,通常为类名 |
+| `traceId` | string | ✅ | 全局唯一追踪ID,从入口生成,全链路透传 |
+| `spanId` | string | 推荐 | 当前节点ID,用于区分调用链中的不同服务 |
+| `userId` | string | 业务必填 | 操作用户标识,未登录可为空 |
+| `message` | string | ✅ | 人类可读的日志摘要,简洁明了 |
+| `fields` | object | ✅ | 结构化业务字段,所有动态数据放入此处 |
+| `file` | string | 推荐 | 代码文件名和行号 |
+| `thread` | string | 推荐 | 线程名 |
+| `host` | string | 推荐 | 主机名或 Pod 名称 |
+| `app` | string | ✅ | 应用名称 |
+| `env` | string | ✅ | dev / test / staging / prod |
+| `error` | object | ERROR时必填 | 包含 `type`、`message`、`stack_trace` |
--- 新增索引
-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 ERROR 日志额外字段
+
+当 `level = ERROR` 时,必须包含:
+
+```json
+{
+ "error": {
+ "type": "ConnectionTimeoutError",
+ "message": "连接下游服务超时",
+ "stack_trace": "完整堆栈信息...",
+ "root_cause": "socket timeout after 3000ms"
+ }
+}
```
-### 3.3 日志中间件设计
+---
+## 四、日志内容规范
+
+### 4.1 日志级别使用标准
+
+| 级别 | 使用场景 | 示例 |
+|------|----------|------|
+| **DEBUG** | 开发调试信息,生产环境**默认关闭** | 变量值、中间计算结果 |
+| **INFO** | 关键业务流程节点、状态变更、外部调用结果 | 订单创建成功、支付回调收到、任务状态变更 |
+| **WARNING** | 可恢复的异常、降级处理、重试、资源使用超阈值 | 重试第3次成功、缓存穿透、磁盘使用率>80% |
+| **ERROR** | 业务失败、系统异常、需要人工介入的错误 | 支付失败、数据库连接断开、第三方接口返回500 |
+
+### 4.2 INFO 级别日志内容要求
+
+每条 INFO 日志必须回答 **5W1H**:
+
+```
+Who(谁操作) + What(做了什么) + When(何时) + Where(哪个服务/节点) + Why(上下文) + How(结果如何)
+```
+
+**✅ 正例:**
```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,
+logger.info(
+ "任务日志拉取成功",
+ extra={
+ "userId": "U10086",
+ "fields": {
+ "jobId": "ft_a016cd8885cd",
+ "tailLines": 5000,
+ "logSize": "2.3MB",
+ "costMs": 42,
+ "source": "frontend"
}
-
- # 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
+logger.info("get logs success")
+logger.info(f"job {job_id} status is {status}") # 禁止字符串拼接
```
-### 3.5 敏感数据脱敏规则
+### 4.3 WARNING/ERROR 日志内容要求
+
+**必须包含三要素**:
+1. 发生了什么(what)
+2. 为什么发生(why)—— 异常类型/错误码
+3. 业务上下文(context)—— 哪个业务对象失败了
+
+**✅ 正例:**
+```python
+logger.warning(
+ "计算轮询检测到任务失败",
+ extra={
+ "fields": {
+ "jobId": "ft_a016cd8885cd",
+ "failureReason": "GPU资源不足",
+ "errorCode": "RESOURCE_INSUFFICIENT",
+ "retryCount": 3,
+ "lastRetryTime": "2026-08-19T08:38:25.495+08:00"
+ }
+ }
+)
+```
+
+**❌ 反例(禁止):**
+```python
+logger.warning("compute polling reported failures") # 没有任何上下文
+logger.error(f"error: {e}") # 只打了异常信息,没有业务ID
+```
+
+### 4.4 禁止打印的内容
+
+| 类别 | 说明 |
+|------|------|
+| 密码/密钥 | 任何形式的 `password`、`secret`、`token`、`api_key` |
+| 个人隐私 | 身份证号、手机号(需脱敏)、银行卡号 |
+| 超大对象 | 超过 1KB 的 JSON/列表/文本内容 |
+| 循环日志 | 禁止在 for/while 循环内打印 INFO 及以上级别 |
+| 异常堆栈重复 | 同一异常在一个请求中只打印一次完整堆栈 |
+
+---
+
+## 五、链路追踪(TraceId)
+
+### 5.1 基本原则
+
+- **入口生成**:网关/前端/定时任务入口生成全局唯一的 `traceId`(32位UUID)
+- **全链路透传**:通过 HTTP Header(`X-Trace-Id`)、RPC Meta、消息队列 Property 向下游传递
+- **日志自动注入**:所有日志输出自动追加 `traceId`,代码中无需手动传入
+- **跨线程传递**:使用 `MDC` 或 `ContextVars` 实现跨线程/协程的透传
+
+### 5.2 实现要求
```python
-# backend/app/core/masking.py
+# Python 示例:使用 logging 的 Filter 自动注入 traceId
+class TraceIdFilter(logging.Filter):
+ def filter(self, record):
+ record.traceId = get_current_trace_id() or "N/A"
+ return True
-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:]}",
-}
+# 所有日志自动带上 traceId
+logger.info("订单创建成功") # 自动注入 traceId,代码无需传参
+```
+**❌ 绝对禁止**:`traceId` 字段值为 `"-"` 或 `null`。
+
+---
+
+## 六、性能与安全
+
+### 6.1 性能要求
+
+| 配置项 | 要求 |
+|--------|------|
+| **异步打印** | 必须使用异步 Appender,禁止同步刷盘阻塞业务线程 |
+| **单文件大小** | ≤ 1GB,达到阈值自动滚动 |
+| **滚动策略** | 按大小滚动(如 1GB)或按天滚动 |
+| **采样率** | 核心业务 100%,非核心(如健康检查、非关键查询)≤ 10% |
+| **禁止打印循环** | 循环体内不得打印 INFO 及以上日志 |
+| **大对象截断** | 超过 1KB 的内容自动截断(前500字符 + 后500字符) |
+
+### 6.2 安全要求
+
+| 要求 | 说明 |
+|------|------|
+| **敏感字段自动脱敏** | 对 `mobile`、`idCard`、`password`、`token` 等字段自动掩码 |
+| **脱敏规则** | 手机号:`138****5678`;身份证:`110***********1234` |
+| **日志查询权限** | 生产日志平台必须有 RBAC 权限控制,禁止随意导出 |
+| **审计追踪** | 谁在什么时候查询了哪些日志,必须记录审计日志 |
+
+### 6.3 脱敏实现示例
+
+```python
+# 脱敏工具函数
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)
+ sensitive_keys = {"password", "token", "api_key", "mobile", "id_card"}
+ for key in sensitive_keys:
+ if key in data:
+ value = str(data[key])
+ if len(value) >= 11: # 手机号
+ data[key] = value[:3] + "****" + value[-4:]
+ elif len(value) >= 18: # 身份证
+ data[key] = value[:3] + "***********" + value[-4:]
return data
```
---
-## 四、实施计划
+## 七、运维与告警
-### 4.1 Phase 1:基础增强(1-2 天)
+### 7.1 日志采集架构
-- [ ] **P1-1** 升级 `JsonLogFormatter`,增加 `trace_id` 字段
-- [ ] **P1-2** 新增 `StructuredLogger` 封装类
-- [ ] **P1-3** 统一所有模块的日志格式为 JSON
-- [ ] **P1-4** 实现 `mask_sensitive()` 脱敏函数
-- [ ] **P1-5** 审计日志表新增 `trace_id`、`duration_ms` 字段
+```
+应用日志(本地文件)
+ ↓
+Filebeat(轻量采集器)
+ ↓
+Kafka(削峰填谷,保证不丢)
+ ↓
+Logstash(解析、过滤、脱敏)
+ ↓
+Elasticsearch(索引存储)
+ ↓
+Kibana / Grafana(查询展示)
+```
-### 4.2 Phase 2:自动化(2-3 天)
+**关键要求**:
+- 禁止应用直接写入 ES,必须经过 Kafka 缓冲
+- Filebeat 采集失败时必须有本地持久化和重试机制
-- [ ] **P2-1** 编写 `@audit_log` 装饰器
-- [ ] **P2-2** 为关键业务接口添加装饰器:
- - 数据集 CRUD
- - 模型 CRUD
- - 微调任务创建/删除
- - 用户登录/登出
- - ACL 授权变更
-- [ ] **P2-3** 实现日志异步写入队列(避免影响性能)
+### 7.2 告警规则
-### 4.3 Phase 3:可观测性(3-5 天)
+| 条件 | 动作 | 优先级 |
+|------|------|--------|
+| 同一服务 5 分钟内出现 ≥ 3 次 ERROR | 钉钉/企微告警 + 电话(P0级) | 最高 |
+| 同一服务 10 分钟内 ERROR 率 > 5% | 钉钉告警(P1级) | 高 |
+| 磁盘使用率 > 80% | 钉钉告警(P2级) | 中 |
+| 单个 ERROR 堆栈重复出现 ≥ 10 次/分钟 | 聚合为一条告警,避免轰炸 | - |
-- [ ] **P3-1** 集成 ELK Stack 或 Loki(可选)
-- [ ] **P3-2** 编写 Grafana 仪表板:
- - 请求量趋势图
- - 错误率统计
- - 慢接口 TOP10
- - 用户操作审计面板
-- [ ] [ ] **P3-3** 实现告警规则(错误率超阈值触发)
+### 7.3 错误聚合策略
+
+- 相同 `error.type` + 相同 `logger` + 相同堆栈前3行 → 视为同一类错误
+- 同一类错误 5 分钟内只发 **1 条告警**(防告警轰炸)
+- 告警内容必须包含:`app`、`env`、`error.type`、首次发生时间、最近发生时间、累计次数
---
-## 五、配置示例
+## 八、日志查询与使用规范
-### 5.1 日志配置 (settings)
+| 场景 | 查询方式 | 时效要求 |
+|------|----------|----------|
+| 日常运维 | Kibana 按 `traceId` 或 `userId` 检索 | 实时 |
+| 异常排查 | 按 `app` + `level:ERROR` + 时间范围 | 实时 |
+| 业务审计 | 按 `userId` + `logger:xxx` + 时间范围 | 30分钟内 |
+| 性能分析 | 按 `costMs` 排序,找出慢请求 | 实时 |
+| 安全审计 | 查询所有访问日志,按 IP/用户筛选 | 按需 |
-```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 # 启用敏感数据脱敏
-```
+## 九、检查清单(Code Review 必查)
-### 5.2 日志输出示例
+| 检查项 | 通过标准 |
+|--------|----------|
+| ☐ JSON 格式 | 所有日志输出均为 JSON,字段名符合规范 |
+| ☐ traceId | 所有日志都有 `traceId`,且不为 `"-"` |
+| ☐ userId | 涉及用户操作的日志都有 `userId` |
+| ☐ 业务上下文 | INFO 日志包含 `fields`,有订单ID/任务ID等 |
+| ☐ ERROR 日志 | 包含 `error.type` + `stack_trace` + 业务ID |
+| ☐ 敏感信息 | 无密码/手机号明文,有脱敏处理 |
+| ☐ 异步打印 | 使用异步 Appender |
+| ☐ 日志级别 | 框架类日志 ≥ WARN,业务日志分级合理 |
+| ☐ 循环内日志 | 无循环内的 INFO 日志 |
+| ☐ 日志分流 | 业务/系统/错误日志分文件存储 |
-**控制台输出(开发环境):**
-```
-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"}
-```
+---
+
+## 十、附:完整日志示例
+
+### 示例一:业务成功流程(INFO)
-**文件输出(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"}}
+{
+ "@timestamp": "2026-08-19T10:30:45.123+08:00",
+ "level": "INFO",
+ "logger": "app.services.job_service",
+ "traceId": "tracer-abc123xyz789",
+ "spanId": "span-001",
+ "userId": "U10086",
+ "message": "计算任务创建成功",
+ "fields": {
+ "jobId": "ft_a016cd8885cd",
+ "jobType": "fine_tuning",
+ "modelId": "model-llama2-7b",
+ "datasetId": "ds-20260819-001",
+ "gpuCount": 4,
+ "estimatedTime": "2h30m",
+ "costMs": 1523,
+ "source": "api"
+ },
+ "file": "job_service.py:234",
+ "thread": "MainThread",
+ "host": "compute-pod-7x9k2",
+ "app": "compute-service",
+ "env": "prod"
+}
```
-**审计日志查询 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;
+### 示例二:可恢复的警告(WARNING)
--- 查询某资源的授权变更历史
-SELECT * FROM audit_logs
-WHERE action LIKE '%acl%'
- AND target_id = 'ds_abc123'
-ORDER BY time DESC;
+```json
+{
+ "@timestamp": "2026-08-19T08:38:27.074+08:00",
+ "level": "WARNING",
+ "logger": "app.workers.compute_poller",
+ "traceId": "tracer-xyz789abc123",
+ "spanId": "span-002",
+ "userId": "U10086",
+ "message": "计算任务状态轮询检测到失败,进入重试",
+ "fields": {
+ "jobId": "ft_a016cd8885cd",
+ "currentStatus": "FAILED",
+ "failureReason": "GPU节点不可用",
+ "errorCode": "NODE_UNAVAILABLE",
+ "retryCount": 2,
+ "maxRetries": 3,
+ "nextRetryDelay": 30
+ },
+ "file": "compute_poller.py:45",
+ "thread": "Thread-8",
+ "host": "compute-pod-7x9k2",
+ "app": "compute-service",
+ "env": "prod"
+}
+```
+
+### 示例三:严重错误(ERROR)+ 告警
+
+```json
+{
+ "@timestamp": "2026-08-19T08:38:30.456+08:00",
+ "level": "ERROR",
+ "logger": "app.services.payment_service",
+ "traceId": "tracer-pay-789xyz",
+ "spanId": "span-003",
+ "userId": "U10086",
+ "message": "支付调用失败,订单状态回滚",
+ "fields": {
+ "orderId": "ORD-20260819-001",
+ "amount": 299.00,
+ "paymentMethod": "wechat",
+ "retryCount": 3,
+ "hasRollback": true
+ },
+ "error": {
+ "type": "PaymentTimeoutException",
+ "message": "支付网关超时,等待响应超过5000ms",
+ "stack_trace": "Traceback (most recent call last):\n File \"payment_service.py:89\" ...",
+ "root_cause": "upstream gateway 10.0.1.100:8080 connection timeout"
+ },
+ "file": "payment_service.py:156",
+ "thread": "http-nio-8080-exec-12",
+ "host": "order-pod-3f8k1",
+ "app": "order-service",
+ "env": "prod"
+}
+```
+
+### 示例四:完整的请求访问日志(ACCESS)
+
+```json
+{
+ "@timestamp": "2026-08-19T10:30:45.001+08:00",
+ "level": "INFO",
+ "logger": "app.middleware.access_log",
+ "traceId": "tracer-abc123xyz789",
+ "userId": "U10086",
+ "message": "HTTP 请求完成",
+ "fields": {
+ "method": "POST",
+ "path": "/api/v1/jobs",
+ "statusCode": 200,
+ "clientIp": "192.168.1.100",
+ "userAgent": "Mozilla/5.0 ...",
+ "requestSize": 2048,
+ "responseSize": 512,
+ "costMs": 1523,
+ "requestBody": {"modelId": "model-llama2-7b", "datasetId": "ds-001"}, // 已脱敏
+ "responseBody": {"jobId": "ft_a016cd8885cd", "status": "CREATED"} // 已脱敏
+ },
+ "app": "compute-service",
+ "env": "prod"
+}
```
---
-## 六、附录
+## 十一、附录:技术栈配置速查
-### A. 日志关键字段说明
+### Python (Logging + JSON)
-| 字段 | 类型 | 说明 | 示例 |
-|------|------|------|------|
-| `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` |
+```python
+import logging
+import json
+from pythonjsonlogger import jsonlogger
-### B. 推荐的 Python 日志库对比
+logger = logging.getLogger("app")
+handler = logging.FileHandler("logs/app-biz.log")
+formatter = jsonlogger.JsonFormatter(
+ fmt="%(asctime)s %(levelname)s %(name)s %(traceId)s %(message)s",
+ rename_fields={"asctime": "@timestamp", "name": "logger"}
+)
+handler.setFormatter(formatter)
+logger.addHandler(handler)
+```
-| 库 | 特点 | 适用场景 |
-|-----|------|---------|
-| `structlog` | 结构化日志,高性能 | 推荐 ✅ |
-| `loguru` | 简单易用,自动配置 | 小型项目 |
-| `logging` | Python 标准库 | 当前已使用 |
+### Java (Logback + JSON)
-### C. 参考链接
+```xml
+
+
大模型微调、评测与推理的一体化工作台