From 78680a5c5cc0775c77388cbac84fb7cfa127a229 Mon Sep 17 00:00:00 2001 From: wangjiming Date: Wed, 19 Aug 2026 17:26:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E7=AB=AF=E6=97=A5?= =?UTF-8?q?=E5=BF=97=EF=BC=8C=E4=B8=94=E6=9B=B4=E6=94=B9=E4=BA=86=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=E5=90=8D=E5=AD=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/v1/endpoints/data_process.py | 20 +- backend/app/api/v1/endpoints/platform.py | 30 + backend/app/core/config.py | 2 +- backend/app/core/logging.py | 519 +++++++++---- backend/app/core/op_log.py | 51 +- backend/app/db/sql/000_full_init.sql | 2 +- compute/api/main.py | 2 +- docker/app/docker-compose.yml | 2 +- docs/governance-user-guide.md | 2 +- docs/security-hardening.md | 2 +- docs/生产级日志系统方案.md | 753 ++++++++++--------- frontend/README.md | 2 +- frontend/index.html | 2 +- frontend/src/components/AppSidebar.vue | 2 +- frontend/src/mock/data.ts | 2 +- frontend/src/router/index.ts | 2 +- frontend/src/views/login/LoginView.vue | 6 +- scripts/verify_platform.py | 2 +- 18 files changed, 865 insertions(+), 538 deletions(-) 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 + + + + + @timestamp + level + thread + logger + + + +``` -- [Python logging cookbook](https://docs.python.org/3/howto/logging.html) -- [ELK Stack 官方文档](https://www.elastic.co/guide/index.html) -- [OpenTelemetry 规范](https://opentelemetry.io/docs/) +### Go (Zap + JSON) + +```go +logger, _ := zap.NewProduction() +logger.Info("订单创建成功", + zap.String("traceId", traceId), + zap.String("userId", userId), + zap.String("orderId", orderId), + zap.Int64("costMs", costMs), +) +``` \ No newline at end of file diff --git a/frontend/README.md b/frontend/README.md index 7220265..dc2571f 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -1,4 +1,4 @@ -# 远光软件微调平台 - 前端（Vue 3） +# 远光智炼 - 前端（Vue 3） > 由原 `web/` 静态多页面 HTML 项目改造而来。 diff --git a/frontend/index.html b/frontend/index.html index 9b525da..6c65e29 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -4,7 +4,7 @@ - 远光软件微调平台 + 远光智炼
diff --git a/frontend/src/components/AppSidebar.vue b/frontend/src/components/AppSidebar.vue index 7d188fb..62df235 100644 --- a/frontend/src/components/AppSidebar.vue +++ b/frontend/src/components/AppSidebar.vue @@ -165,7 +165,7 @@ async function handleLogout() { diff --git a/frontend/src/mock/data.ts b/frontend/src/mock/data.ts index 78d6544..65bb884 100644 --- a/frontend/src/mock/data.ts +++ b/frontend/src/mock/data.ts @@ -320,7 +320,7 @@ export const mockDatasetPreviews: Record = { '{"instruction":"请概括以下文本的核心观点","input":"人工智能正在提升企业的数据处理效率。","output":"人工智能能够帮助企业提升数据处理效率。"}', '{"instruction":"将用户问题改写为更清晰的表达","input":"这个功能咋用？","output":"请说明该功能的具体使用步骤。"}', '{"instruction":"判断文本情感倾向","input":"这次服务响应很及时，问题也解决了。","output":"正向"}', - '{"instruction":"提取文本中的关键实体","input":"远光软件于周一发布了新的模型管理平台。","output":["远光软件","周一","模型管理平台"]}', + '{"instruction":"提取文本中的关键实体","input":"远光智炼于周一发布了新的模型管理平台。","output":["远光智炼","周一","模型管理平台"]}', '{"instruction":"生成简短回复","input":"您好，我想了解数据集上传支持哪些格式？","output":"您好，目前支持 JSON、JSONL、CSV、TXT 等常见格式。"}', '{"instruction":"对以下内容进行分类","input":"如何重置账户密码？","output":"账户与安全"}', '{"instruction":"找出句子中的时间信息","input":"系统将在 7 月 15 日凌晨 2 点进行升级。","output":"7 月 15 日凌晨 2 点"}', diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 710987a..973a062 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -383,7 +383,7 @@ function requiredPermission(path: string, explicit?: unknown) { router.beforeEach((to, _from, next) => { if (!to.meta.public) routeLoading.value = true const auth = useAuthStore() - document.title = to.meta.title ? `${to.meta.title} - 远光软件微调平台` : '远光软件微调平台' + document.title = to.meta.title ? `${to.meta.title} - 远光智炼` : '远光智炼' if (to.meta.public) { // 已登录访问登录页则跳转主页 diff --git a/frontend/src/views/login/LoginView.vue b/frontend/src/views/login/LoginView.vue index a13c282..d9cf556 100644 --- a/frontend/src/views/login/LoginView.vue +++ b/frontend/src/views/login/LoginView.vue @@ -43,7 +43,7 @@ async function handleLogin() {
-
diff --git a/scripts/verify_platform.py b/scripts/verify_platform.py index 977b5d5..64fc0e2 100644 --- a/scripts/verify_platform.py +++ b/scripts/verify_platform.py @@ -178,7 +178,7 @@ def run_training_smoke(backend_url: str) -> dict[str, Any]: def main() -> int: - parser = argparse.ArgumentParser(description="Verify YG Fine-Tune platform deployment.") + parser = argparse.ArgumentParser(description="Verify YG Zhilian deployment.") parser.add_argument("--frontend-url", default="http://localhost:16801", help="Frontend base URL.") parser.add_argument("--backend-url", default="http://localhost:17861/modelTF", help="Backend API base URL with /modelTF.") parser.add_argument("--username", default="admin", help="Login username.")