diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py
index 0b4c052..970a0ef 100644
--- a/backend/app/api/v1/endpoints/data_process.py
+++ b/backend/app/api/v1/endpoints/data_process.py
@@ -41,8 +41,12 @@ from app.core.auth import (
has_resource_access,
is_admin,
)
+
+from app.core.logging import get_structured_logger
+
from app.core.config import get_settings
from app.db.platform_store import get_platform_store
+
from app.modules.data_process.algorithms import (
ParsedText,
canonical_record_json,
@@ -117,6 +121,7 @@ from app.schemas.data_process import (
)
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
@@ -292,7 +297,7 @@ def _commit_source_batch(
except Exception:
# 文件系统回滚失败不能覆盖数据库抛出的根因,并继续清理其余对象。
logger.exception(
- "failed to roll back data process source object task_id=%s",
+ "数据处理源对象回滚失败 task_id=%s",
task_id,
)
raise
@@ -662,7 +667,7 @@ def _run_generation(
) -> None:
started_at = time.perf_counter()
logger.info(
- "data process generation worker started task_id=%s generation_run_id=%s",
+ "数据处理生成任务开始 task_id=%s generation_run_id=%s",
task_id,
generation_run_id,
)
@@ -670,8 +675,7 @@ def _run_generation(
task = store.get_task(task_id)
if not store.generation_is_running(task_id, generation_run_id):
logger.info(
- "data process generation worker skipped inactive run task_id=%s "
- "generation_run_id=%s",
+ "数据处理生成任务跳过(非活跃运行) task_id=%s generation_run_id=%s",
task_id,
generation_run_id,
)
@@ -761,8 +765,7 @@ def _run_generation(
len(preview_items),
):
logger.info(
- "data process generation stopped before completion task_id=%s "
- "generation_run_id=%s",
+ "数据处理生成任务被中止 task_id=%s generation_run_id=%s",
task_id,
generation_run_id,
)
@@ -853,9 +856,7 @@ def _run_generation(
"created_by": (store.get_task(task_id) or {}).get("created_by"),
})
logger.info(
- "data process generation completed task_id=%s generation_run_id=%s "
- "output_count=%s filtered_count=%s duplicate_count=%s error_count=%s "
- "duration_ms=%.2f",
+ "数据处理生成完成 task_id=%s generation_run_id=%s output_count=%s filtered_count=%s duplicate_count=%s error_count=%s duration_ms=%.2f",
task_id,
generation_run_id,
completed.get("output_count", len(accepted)),
@@ -866,14 +867,13 @@ def _run_generation(
)
else:
logger.info(
- "data process generation stopped before result persistence task_id=%s "
- "generation_run_id=%s",
+ "数据处理生成任务在持久化前被停止 task_id=%s generation_run_id=%s",
task_id,
generation_run_id,
)
except Exception as exc:
logger.exception(
- "data process generation failed task_id=%s generation_run_id=%s duration_ms=%.2f",
+ "数据处理生成失败 task_id=%s generation_run_id=%s duration_ms=%.2f",
task_id,
generation_run_id,
(time.perf_counter() - started_at) * 1000,
@@ -887,8 +887,7 @@ def _run_generation(
)
except Exception:
logger.exception(
- "failed to persist data process generation failure task_id=%s "
- "generation_run_id=%s",
+ "数据处理生成失败持久化异常 task_id=%s generation_run_id=%s",
task_id,
generation_run_id,
)
@@ -943,6 +942,7 @@ def create_task(
values["tenant_id"] = current_user.get("tenant_id") or "default"
get_platform_store().assert_active_tenant(values["tenant_id"])
task = store.create_task(values)
+ biz_logger.info("用户创建数据处理任务成功", taskId=task["id"], processType=task.get("process_type", ""))
return ok(task, "data process task created")
@@ -968,10 +968,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("用户更新数据处理任务成功", taskId=task_id)
+ return ok(result, "data process task updated")
@router.put("/{task_id}/workflow-step")
@@ -1060,7 +1059,7 @@ def _remove_repeated_storage_objects(
)
except Exception:
logger.exception(
- "failed to roll back repeated data process source object task_id=%s",
+ "数据处理源对象重复回滚失败 task_id=%s",
task_id,
)
@@ -1136,6 +1135,7 @@ def delete_task(
) -> dict[str, Any]:
with api_errors():
store.delete_task(task_id)
+ biz_logger.info("用户删除数据处理任务成功", taskId=task_id)
return ok({"deleted": task_id}, "data process task deleted")
@@ -1448,13 +1448,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("用户删除数据处理源文件成功", taskId=task_id, fileId=file_id, storageCleanupPending=cleanup_pending)
return ok(
{"deleted": file_id, "storage_cleanup_pending": cleanup_pending},
"source file removed",
@@ -1727,7 +1726,7 @@ def _prepare_preview_items(
extracted_text = "\n\n".join(page.text for page in pages if page.text)
if extracted_text != str(source.get("content") or ""):
logger.warning(
- "skip PDF document noise detection because stored offsets differ for %s",
+ "跳过PDF文档噪声检测(存储偏移量不一致) source_id=%s",
source["id"],
)
continue
@@ -1750,7 +1749,7 @@ def _run_preview(
started_at = time.perf_counter()
logger.info(
- "data process preview started task_id=%s preview_run_id=%s total_files=%s",
+ "数据处理预览开始 task_id=%s preview_run_id=%s total_files=%s",
task_id,
preview_run_id,
len(source_file_ids),
@@ -1759,7 +1758,7 @@ def _run_preview(
is_unstructured = store.get_task(task_id).get("process_type") == "unstructured"
if not store.mark_preview_running(task_id, preview_run_id):
logger.info(
- "data process preview skipped inactive run task_id=%s preview_run_id=%s",
+ "数据处理预览跳过(非活跃运行) task_id=%s preview_run_id=%s",
task_id,
preview_run_id,
)
@@ -1769,8 +1768,7 @@ def _run_preview(
for completed_files, source_file_id in enumerate(source_file_ids, start=1):
if not store.preview_is_running(task_id, preview_run_id):
logger.info(
- "data process preview cancelled task_id=%s preview_run_id=%s "
- "completed_files=%s total_files=%s",
+ "数据处理预览被取消 task_id=%s preview_run_id=%s completed_files=%s total_files=%s",
task_id,
preview_run_id,
completed_files - 1,
@@ -1801,8 +1799,7 @@ def _run_preview(
total_files,
):
logger.info(
- "data process preview stopped before progress update task_id=%s "
- "preview_run_id=%s completed_files=%s total_files=%s",
+ "数据处理预览在进度更新前被停止 task_id=%s preview_run_id=%s completed_files=%s total_files=%s",
task_id,
preview_run_id,
completed_files,
@@ -1811,8 +1808,7 @@ def _run_preview(
return
if store.complete_preview(task_id, preview_run_id):
logger.info(
- "data process preview completed task_id=%s preview_run_id=%s "
- "total_files=%s total_items=%s duration_ms=%.2f",
+ "数据处理预览完成 task_id=%s preview_run_id=%s total_files=%s total_items=%s duration_ms=%.2f",
task_id,
preview_run_id,
total_files,
@@ -1821,14 +1817,13 @@ def _run_preview(
)
else:
logger.info(
- "data process preview completion ignored for inactive run task_id=%s "
- "preview_run_id=%s",
+ "数据处理预览完成但运行已失效 task_id=%s preview_run_id=%s",
task_id,
preview_run_id,
)
except Exception as exc:
logger.exception(
- "data process preview failed task_id=%s preview_run_id=%s duration_ms=%.2f",
+ "数据处理预览失败 task_id=%s preview_run_id=%s duration_ms=%.2f",
task_id,
preview_run_id,
(time.perf_counter() - started_at) * 1000,
@@ -1842,8 +1837,7 @@ def _run_preview(
)
except Exception:
logger.exception(
- "failed to persist data process preview failure task_id=%s "
- "preview_run_id=%s",
+ "数据处理预览失败持久化异常 task_id=%s preview_run_id=%s",
task_id,
preview_run_id,
)
@@ -2053,6 +2047,7 @@ def stop(
) -> dict[str, Any]:
with api_errors():
store.stop_task(task_id)
+ biz_logger.info("用户停止数据处理任务成功", taskId=task_id)
return ok(store.progress(task_id), "data process task stopped")
@@ -2094,7 +2089,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("用户确认数据处理结果成功", taskId=task_id)
+ return ok(result, "data process results confirmed")
@router.put("/{task_id}/results/{result_id}")
@@ -2443,8 +2440,7 @@ def regenerate_results_batch(
}))
logger.info(
- "data process result batch regeneration started batch_id=%s task_id=%s "
- "requested=%s prepared=%s concurrency=%s",
+ "数据处理结果批量重新生成开始 batch_id=%s task_id=%s requested=%s prepared=%s concurrency=%s",
batch_id,
task_id,
len(payload.items),
@@ -2512,8 +2508,7 @@ def regenerate_results_batch(
except Exception as exc: # pragma: no cover - defensive boundary
outcome = "internal_error"
logger.exception(
- "data process result batch regeneration crashed "
- "batch_id=%s task_id=%s result_id=%s",
+ "数据处理结果批量重新生成崩溃 batch_id=%s task_id=%s result_id=%s",
batch_id,
task_id,
result_id,
@@ -2524,8 +2519,7 @@ def regenerate_results_batch(
"message": _safe_regeneration_error(exc),
}))
logger.info(
- "data process result batch item finished batch_id=%s task_id=%s "
- "result_id=%s outcome=%s duration_ms=%.2f",
+ "数据处理结果批量项完成 batch_id=%s task_id=%s result_id=%s outcome=%s duration_ms=%.2f",
batch_id,
task_id,
result_id,
@@ -2545,8 +2539,7 @@ def regenerate_results_batch(
)
duration_ms = (time.perf_counter() - started_at) * 1000
logger.info(
- "data process result batch regeneration completed batch_id=%s task_id=%s "
- "succeeded=%s failed=%s remaining_invalid=%s duration_ms=%.2f",
+ "数据处理结果批量重新生成完成 batch_id=%s task_id=%s succeeded=%s failed=%s remaining_invalid=%s duration_ms=%.2f",
batch_id,
task_id,
len(success_items),
@@ -2593,8 +2586,7 @@ def evaluate_results_batch(
evaluation_model = store.get_generation_model(str(model_id))
except NotFoundError:
logger.warning(
- "data process evaluation model unavailable, judge layer "
- "skipped task_id=%s model_id=%s",
+ "数据处理评测模型不可用,跳过评测层 task_id=%s model_id=%s",
task_id,
model_id,
)
@@ -2642,8 +2634,7 @@ def evaluate_results_batch(
}))
logger.info(
- "data process result batch evaluation started batch_id=%s task_id=%s "
- "requested=%s prepared=%s judge_enabled=%s",
+ "数据处理结果批量评测开始 batch_id=%s task_id=%s requested=%s prepared=%s judge_enabled=%s",
batch_id,
task_id,
len(payload.items),
@@ -2660,8 +2651,7 @@ def evaluate_results_batch(
semantic_embedding_model()
except Exception:
logger.warning(
- "data process semantic embedding unavailable, semantic layer "
- "will be skipped batch_id=%s",
+ "数据处理语义嵌入模型不可用,语义层将跳过 batch_id=%s",
batch_id,
)
request_timeout = _result_regeneration_timeout(config)
@@ -2714,8 +2704,7 @@ def evaluate_results_batch(
"message": _safe_regeneration_error(exc),
}))
logger.info(
- "data process result batch evaluation item finished "
- "batch_id=%s task_id=%s result_id=%s outcome=%s duration_ms=%.2f",
+ "数据处理结果批量评测项完成 batch_id=%s task_id=%s result_id=%s outcome=%s duration_ms=%.2f",
batch_id,
task_id,
result_id,
@@ -2727,8 +2716,7 @@ def evaluate_results_batch(
failure_items = [item for _, item in sorted(failures, key=lambda pair: pair[0])]
duration_ms = (time.perf_counter() - started_at) * 1000
logger.info(
- "data process result batch evaluation completed batch_id=%s task_id=%s "
- "succeeded=%s failed=%s duration_ms=%.2f",
+ "数据处理结果批量评测完成 batch_id=%s task_id=%s succeeded=%s failed=%s duration_ms=%.2f",
batch_id,
task_id,
len(success_items),
@@ -2784,8 +2772,7 @@ def regenerate_result(
)
except _ResultRegenerationFailed as exc:
logger.warning(
- "data process result regeneration failed task_id=%s result_id=%s "
- "duration_ms=%.2f reason=%s",
+ "数据处理结果重新生成失败 task_id=%s result_id=%s duration_ms=%.2f reason=%s",
task_id,
result_id,
(time.perf_counter() - started_at) * 1000,
@@ -2793,7 +2780,7 @@ def regenerate_result(
)
raise
logger.info(
- "data process result regenerated task_id=%s result_id=%s duration_ms=%.2f",
+ "数据处理结果重新生成完成 task_id=%s result_id=%s duration_ms=%.2f",
task_id,
result_id,
(time.perf_counter() - started_at) * 1000,
@@ -2809,5 +2796,6 @@ def publish(
) -> dict[str, Any]:
with api_errors():
result = store.publish(task_id, payload.model_dump(mode="json"))
+ biz_logger.info("用户发布数据处理任务成功", 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 d7dc726..168d9c7 100644
--- a/backend/app/api/v1/endpoints/platform.py
+++ b/backend/app/api/v1/endpoints/platform.py
@@ -913,6 +913,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"
@@ -930,6 +931,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(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
session_id = payload.get("session_id", "")
@@ -1179,6 +1181,7 @@ async def users(current_user: dict = Depends(require_admin)) -> 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(...), current_user: dict = Depends(require_admin)) -> dict[str, Any]:
payload = dict(payload)
payload.setdefault("password", "123456")
@@ -1187,6 +1190,7 @@ async def create_user(payload: dict[str, Any] = Body(...), current_user: dict =
@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(...), current_user: dict = Depends(require_admin)) -> dict[str, Any]:
try:
return ok(get_platform_store().update_user(user_id, payload))
@@ -1195,6 +1199,7 @@ async def update_user(user_id: str, payload: dict[str, Any] = Body(...), current
@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), current_user: dict = Depends(require_admin)) -> dict[str, Any]:
try:
get_platform_store().delete_user(user_id, deleted_by=str(current_user.get("id") or "system"))
@@ -1212,6 +1217,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={}),
@@ -1228,6 +1234,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),
@@ -1310,6 +1317,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")
@@ -1424,6 +1432,7 @@ async def test_online_model(payload: dict[str, Any] = Body(...), current_user: d
@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",
@@ -1465,6 +1474,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",
@@ -1481,6 +1491,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):
@@ -1492,6 +1503,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):
@@ -1691,6 +1703,7 @@ async def dataset_version_content(file_id: str, version_id: str, current_user: d
@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(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
try:
store = get_platform_store()
@@ -1715,6 +1728,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, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
try:
store = get_platform_store()
@@ -1913,6 +1927,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=[]),
@@ -2080,6 +2095,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",
@@ -2110,6 +2126,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",
@@ -2175,6 +2192,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",
@@ -2418,6 +2436,7 @@ async def fine_tune_gpu_status(task_id: str, current_user: dict[str, Any] = Depe
@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")
@@ -2452,6 +2471,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 {}
@@ -2883,6 +2903,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")
@@ -2902,6 +2923,7 @@ async def dimension_list(current_user: dict = Depends(get_current_user)) -> dict
@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(...), current_user: dict = Depends(require_admin)) -> dict[str, Any]:
return ok(get_platform_store().create_dimension(payload))
@@ -2915,6 +2937,7 @@ async def dimension_detail(dimension_id: str, current_user: dict = Depends(get_c
@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(...), current_user: dict = Depends(require_admin)) -> dict[str, Any]:
try:
return ok(get_platform_store().update_dimension(dimension_id, payload))
@@ -2923,6 +2946,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, current_user: dict = Depends(require_admin)) -> dict[str, Any]:
get_platform_store().delete_dimension(dimension_id)
return ok({"deleted": dimension_id})
@@ -2943,6 +2967,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]:
bind_active_tenant(payload, current_user)
payload.setdefault("created_by", current_user.get("id"))
@@ -3785,6 +3810,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")
@@ -3797,6 +3823,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")
@@ -3809,6 +3836,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")
@@ -4123,6 +4151,7 @@ async def compute_job_detail(job_id: str, current_user: dict = Depends(get_curre
@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"):
@@ -4166,6 +4195,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 8b75042..285d85c 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 bf28341..5aa7e8d 100644
--- a/backend/app/core/logging.py
+++ b/backend/app/core/logging.py
@@ -1,22 +1,26 @@
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="")
@@ -30,54 +34,74 @@ def get_client_ip(request: Request | None) -> str:
return value.split(",", 1)[0].strip()
return request.client.host if request.client else ""
-# ==================== 敏感数据脱敏规则 ====================
+# ==================== 敏感数据脱敏 ====================
-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
-
# Mask the value as well as the key. Replacing only ``api_key=`` would
# still leak the credential in audit messages and exception text.
assignment_pattern = (
@@ -90,92 +114,165 @@ def mask_sensitive_string(text: str) -> str:
except re.error:
pass
- patterns = [
- (r'Bearer\s+[A-Za-z0-9\-._]+', 'Bearer ***'),
- (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,
@@ -233,10 +330,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)
@@ -251,11 +346,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
@@ -269,7 +362,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$"
@@ -283,99 +375,126 @@ 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)
ip_token = client_ip_var.set(get_client_ip(request))
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
+
+ # 结构化访问日志(中文 message,方便直接阅读)
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 "-",
+ f"HTTP请求 {request.method} {request.url.path} → {response.status_code}(耗时{round(elapsed_ms, 2)}ms)",
+ 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
@@ -393,18 +512,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(
+ f"HTTP请求异常 {request.method} {request.url.path}(耗时{round(elapsed_ms, 2)}ms)— 服务内部错误",
+ extra={
+ "request_method": request.method,
+ "request_path": request.url.path,
+ "duration_ms": round(elapsed_ms, 2),
+ "client_ip": client_ip,
+ },
+ exc_info=True,
)
# 未被捕获的异常,写入操作日志
@@ -426,7 +550,7 @@ def setup_request_logging(app: FastAPI) -> None:
duration_ms=elapsed_ms,
)
except Exception:
- pass # 日志写入失败不影响主流程
+ pass
raise
finally:
@@ -437,53 +561,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 970b2d1..24c7751 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_client_ip, 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])
@@ -76,6 +77,92 @@ class OpStatus:
FAILURE = "failure"
+# ==================== 中文映射表(让日志 message 直接可读)====================
+
+MODULE_CN: dict[str, str] = {
+ "fine-tune": "模型训练",
+ "model-eval": "模型评测",
+ "model-inference": "模型推理",
+ "model-manage": "模型管理",
+ "dataset": "数据集",
+ "data-process": "数据处理",
+ "data-convert": "数据转换",
+ "compute": "算力节点",
+ "system": "系统",
+}
+
+ACTION_CN: dict[str, str] = {
+ "create": "创建",
+ "update": "更新",
+ "delete": "删除",
+ "start": "启动",
+ "stop": "停止",
+ "upload": "上传",
+ "download": "下载",
+ "convert": "转换",
+ "merge": "合并",
+ "import": "导入",
+ "login": "登录",
+ "logout": "退出登录",
+ "publish": "发布",
+ "retry": "重试",
+ "request": "请求",
+}
+
+TARGET_TYPE_CN: dict[str, str] = {
+ "fine_tune": "训练任务",
+ "eval": "评测任务",
+ "inference": "推理任务",
+ "model": "模型",
+ "trained_model": "训练产出模型",
+ "dataset": "数据集",
+ "dataset_version": "数据集版本",
+ "data_process_task": "数据处理任务",
+ "user": "用户",
+ "api": "接口",
+}
+
+
+def _build_cn_message(
+ module: str,
+ action: str,
+ target_type: str,
+ target_name: str | None,
+ target_id: str | None,
+ status: str,
+ username: str | None,
+ error_type: str,
+ error_message: str,
+) -> str:
+ """构建中文人类可读的日志消息,格式:[用户] 对 [模块] 执行了 [动作],结果:成功/失败"""
+ user_part = f"用户[{username}]" if username else "系统"
+ module_cn = MODULE_CN.get(module, module)
+ action_cn = ACTION_CN.get(action, action)
+ target_cn = TARGET_TYPE_CN.get(target_type, target_type or "")
+ target_label = target_name or target_id or ""
+
+ # 拼接操作对象描述
+ if target_cn and target_label:
+ target_part = f"{target_cn}「{target_label}」"
+ elif target_cn:
+ target_part = target_cn
+ elif target_label:
+ target_part = f"「{target_label}」"
+ else:
+ target_part = ""
+
+ if status == OpStatus.SUCCESS:
+ result = "成功"
+ msg = f"{user_part} {action_cn}{module_cn}{target_part},结果:成功"
+ else:
+ result = "失败"
+ err_brief = error_message[:120] if error_message else ""
+ err_part = f"({error_type}: {err_brief})" if error_type and err_brief else f"({error_type})" if error_type else ""
+ msg = f"{user_part} {action_cn}{module_cn}{target_part},结果:失败{err_part}"
+
+ return msg
+
+
def op_log(
module: str,
action: str,
@@ -341,20 +428,54 @@ 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)----
+ cn_msg = _build_cn_message(
+ module=module, action=action, target_type=target_type,
+ target_name=target_name, target_id=target_id, status=status,
+ username=username, error_type=error_type, error_message=error_message,
+ )
+
+ 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 req_method:
+ log_fields["requestMethod"] = req_method
+ if req_path:
+ log_fields["requestPath"] = req_path
+ 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(cn_msg, **log_fields)
+ elif status == OpStatus.FAILURE:
+ biz_logger.error(cn_msg, **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 = get_client_ip(request) or None
- req_method = request.method
- req_path = request.url.path
-
with store.connect() as conn:
conn.execute(
"""
@@ -379,4 +500,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 441139f..81d60a6 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/backend/app/workers/compute_poller.py b/backend/app/workers/compute_poller.py
index 6b68f8a..e041068 100644
--- a/backend/app/workers/compute_poller.py
+++ b/backend/app/workers/compute_poller.py
@@ -15,11 +15,11 @@ logger = get_logger(__name__)
async def run_compute_poller() -> None:
settings = get_settings()
if settings.compute_mode == "simulator" or settings.compute_status_sync_mode != "polling":
- logger.info("compute poller disabled", extra={"compute_mode": settings.compute_mode})
+ logger.info("计算轮询已禁用", extra={"compute_mode": settings.compute_mode})
return
interval = max(3, settings.compute_poll_interval_seconds)
- logger.info("compute poller started", extra={"interval_seconds": interval})
+ logger.info("计算轮询已启动", extra={"interval_seconds": interval})
# PlatformStore may run additive schema checks against a remote PostgreSQL
# server on first use. Keep that startup work off the Uvicorn event loop so
# health checks and normal API requests can still respond while the DB is
@@ -41,19 +41,19 @@ async def run_compute_poller() -> None:
now = time.monotonic()
if signature != last_failure_signature or now - last_failure_logged_at >= 300:
logger.warning(
- "compute polling reported failures count=%d first_error=%s",
+ "计算轮询报告失败任务 count=%d first_error=%s",
len(result["failed"]),
signature[:500],
)
last_failure_signature = signature
last_failure_logged_at = now
elif result["synced"]:
- logger.debug("compute jobs synchronized", extra={"result": result})
+ logger.debug("计算任务状态已同步", extra={"result": result})
except asyncio.CancelledError:
- logger.info("compute poller stopped")
+ logger.info("计算轮询已停止")
raise
except Exception as exc: # noqa: BLE001 - keep background polling alive
- logger.exception("compute poller failed", extra={"error": str(exc)})
+ logger.exception("计算轮询执行失败", extra={"error": str(exc)})
if "store" in locals() and isinstance(exc, (ConnectionError, TimeoutError)):
store = None
await asyncio.sleep(interval)
diff --git a/backend/test_results.json b/backend/test_results.json
new file mode 100644
index 0000000..8a68cb0
--- /dev/null
+++ b/backend/test_results.json
@@ -0,0 +1,266 @@
+[
+ {
+ "name": "health",
+ "method": "GET",
+ "url": "/health",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(4 keys)"
+ },
+ {
+ "name": "system-info",
+ "method": "GET",
+ "url": "/system-info",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(7 keys)"
+ },
+ {
+ "name": "me",
+ "method": "GET",
+ "url": "/me",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(9 keys)"
+ },
+ {
+ "name": "dashboard/overview",
+ "method": "GET",
+ "url": "/dashboard/overview",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(6 keys)"
+ },
+ {
+ "name": "dashboard/stats",
+ "method": "GET",
+ "url": "/dashboard/stats",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(9 keys)"
+ },
+ {
+ "name": "users-list",
+ "method": "GET",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "9 items"
+ },
+ {
+ "name": "users-create",
+ "method": "POST",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(9 keys)"
+ },
+ {
+ "name": "users-change-password",
+ "method": "POST",
+ "url": "/users/me/password",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "model-manage-list",
+ "method": "GET",
+ "url": "/model-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "5 items"
+ },
+ {
+ "name": "model-manage-local",
+ "method": "GET",
+ "url": "/model-manage/local-models",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "model-manage-trained",
+ "method": "GET",
+ "url": "/model-manage/trained-models",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "model-manage-export-jobs",
+ "method": "GET",
+ "url": "/model-manage/export-jobs",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "11 items"
+ },
+ {
+ "name": "model-manage-create",
+ "method": "POST",
+ "url": "/model-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(17 keys)"
+ },
+ {
+ "name": "dataset-list",
+ "method": "GET",
+ "url": "/dataset-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "29 items"
+ },
+ {
+ "name": "dataset-create",
+ "method": "POST",
+ "url": "/dataset-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "fine-tune-list",
+ "method": "GET",
+ "url": "/fine-tune",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "9 items"
+ },
+ {
+ "name": "fine-tune-check-name",
+ "method": "GET",
+ "url": "/fine-tune/check-name?name=test_task",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "fine-tune-preflight",
+ "method": "POST",
+ "url": "/fine-tune/preflight",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(4 keys)"
+ },
+ {
+ "name": "model-eval-list",
+ "method": "GET",
+ "url": "/model-eval",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "4 items"
+ },
+ {
+ "name": "dimension-list",
+ "method": "GET",
+ "url": "/dimension",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "18 items"
+ },
+ {
+ "name": "model-compare-list",
+ "method": "GET",
+ "url": "/model-compare",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "6 items"
+ },
+ {
+ "name": "model-chat-local-status",
+ "method": "GET",
+ "url": "/model-chat/local/status",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(8 keys)"
+ },
+ {
+ "name": "data-process-list",
+ "method": "GET",
+ "url": "/data-process",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(4 keys)"
+ },
+ {
+ "name": "compute-nodes",
+ "method": "GET",
+ "url": "/compute/nodes",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "2 items"
+ },
+ {
+ "name": "compute-gpus",
+ "method": "GET",
+ "url": "/compute/gpus",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "2 items"
+ },
+ {
+ "name": "compute-queue",
+ "method": "GET",
+ "url": "/compute/queue",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "0 items"
+ },
+ {
+ "name": "log-files",
+ "method": "GET",
+ "url": "/log-files",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "2 items"
+ },
+ {
+ "name": "training-log-files",
+ "method": "GET",
+ "url": "/training-log-files",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "9 items"
+ },
+ {
+ "name": "web-log",
+ "method": "POST",
+ "url": "/web-log",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(3 keys)"
+ },
+ {
+ "name": "data-convert-list",
+ "method": "GET",
+ "url": "/data-convert",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(2 keys)"
+ },
+ {
+ "name": "error-404",
+ "method": "GET",
+ "url": "/nonexistent-endpoint",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "error-unauthorized",
+ "method": "GET",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "10 items"
+ },
+ {
+ "name": "viewer-login",
+ "method": "POST",
+ "url": "/login",
+ "status": "SKIP",
+ "message": "viewer user not found",
+ "data_desc": "-"
+ }
+]
\ No newline at end of file
diff --git a/backend/test_results_advanced.json b/backend/test_results_advanced.json
new file mode 100644
index 0000000..5dd2019
--- /dev/null
+++ b/backend/test_results_advanced.json
@@ -0,0 +1,258 @@
+[
+ {
+ "name": "crud-model-create",
+ "method": "POST",
+ "url": "/model-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(17 keys)"
+ },
+ {
+ "name": "crud-model-get-by-id",
+ "method": "GET",
+ "url": "/model-manage/m_dcebe627d644",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(17 keys)"
+ },
+ {
+ "name": "crud-model-update",
+ "method": "PUT",
+ "url": "/model-manage/m_dcebe627d644",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(17 keys)"
+ },
+ {
+ "name": "crud-model-purpose",
+ "method": "PUT",
+ "url": "/model-manage/m_dcebe627d644/purpose",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(17 keys)"
+ },
+ {
+ "name": "crud-model-delete",
+ "method": "DELETE",
+ "url": "/model-manage/m_dcebe627d644",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "crud-dataset-create",
+ "method": "POST",
+ "url": "/dataset-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "crud-dataset-get-by-id",
+ "method": "GET",
+ "url": "/dataset-manage/ds_b8dd915d5e09",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(28 keys)"
+ },
+ {
+ "name": "crud-dataset-update",
+ "method": "PUT",
+ "url": "/dataset-manage/ds_b8dd915d5e09",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(27 keys)"
+ },
+ {
+ "name": "crud-dataset-delete",
+ "method": "DELETE",
+ "url": "/dataset-manage/ds_b8dd915d5e09",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "crud-user-create",
+ "method": "POST",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(9 keys)"
+ },
+ {
+ "name": "crud-user-list",
+ "method": "GET",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "11 items"
+ },
+ {
+ "name": "crud-user-update",
+ "method": "PUT",
+ "url": "/users/u_eb94ee60769e",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(9 keys)"
+ },
+ {
+ "name": "crud-user-reset-pwd",
+ "method": "POST",
+ "url": "/users/u_eb94ee60769e/reset-password",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "crud-user-delete",
+ "method": "DELETE",
+ "url": "/users/u_eb94ee60769e",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(2 keys)"
+ },
+ {
+ "name": "error-invalid-model-id",
+ "method": "GET",
+ "url": "/model-manage/nonexistent_id_12345",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "error-invalid-dataset-id",
+ "method": "GET",
+ "url": "/dataset-manage/nonexistent_id_12345",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "error-invalid-finetune-id",
+ "method": "GET",
+ "url": "/fine-tune/nonexistent_id_12345",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "error-invalid-eval-id",
+ "method": "GET",
+ "url": "/model-eval/nonexistent_id_12345",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "error-duplicate-login",
+ "method": "POST",
+ "url": "/login",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "error-missing-fields",
+ "method": "POST",
+ "url": "/model-manage",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(17 keys)"
+ },
+ {
+ "name": "auth-no-token-users",
+ "method": "GET",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "10 items"
+ },
+ {
+ "name": "auth-no-token-finetune",
+ "method": "GET",
+ "url": "/fine-tune",
+ "status": "FAIL(code=-1)",
+ "message": "",
+ "data_desc": "null"
+ },
+ {
+ "name": "auth-invalid-token",
+ "method": "GET",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "10 items"
+ },
+ {
+ "name": "auth-empty-token",
+ "method": "GET",
+ "url": "/users",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "10 items"
+ },
+ {
+ "name": "crud-dimension-create",
+ "method": "POST",
+ "url": "/dimension",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(6 keys)"
+ },
+ {
+ "name": "crud-dimension-get",
+ "method": "GET",
+ "url": "/dimension/dim_12124ed44bbe",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(6 keys)"
+ },
+ {
+ "name": "crud-dimension-update",
+ "method": "PUT",
+ "url": "/dimension/dim_12124ed44bbe",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(6 keys)"
+ },
+ {
+ "name": "crud-dimension-delete",
+ "method": "DELETE",
+ "url": "/dimension/dim_12124ed44bbe",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(1 keys)"
+ },
+ {
+ "name": "compute-nodes-detail",
+ "method": "GET",
+ "url": "/compute/nodes",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "2 items"
+ },
+ {
+ "name": "compute-nodes-list2",
+ "method": "GET",
+ "url": "/compute/nodes",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "2 items"
+ },
+ {
+ "name": "compute-node-replicas",
+ "method": "GET",
+ "url": "/compute/nodes/node_1499a71b4871/replicas",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "9 items"
+ },
+ {
+ "name": "compute-node-engines",
+ "method": "GET",
+ "url": "/compute/nodes/node_1499a71b4871/engines",
+ "status": "PASS",
+ "message": "ok",
+ "data_desc": "obj(2 keys)"
+ }
+]
\ No newline at end of file
diff --git a/compute/api/main.py b/compute/api/main.py
index 2bbd91d..4580be6 100644
--- a/compute/api/main.py
+++ b/compute/api/main.py
@@ -24,7 +24,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 2711833..d5cf0db 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 5e7d72b..1288c33 100644
--- a/docs/governance-user-guide.md
+++ b/docs/governance-user-guide.md
@@ -1,5 +1,6 @@
# 平台治理功能使用指南
+
> 版本:v1.3
> 日期:2026-08-19
> 适用版本:YG Fine-Tune Platform 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
+
+
大模型微调、评测与推理的一体化工作台