fix: 限制日志暴涨并修复启动日志兼容问题

_public/static/admin/js/config.js

@@ -14,6 +14,8 @@ const NUMERIC_FIELDS = new Set([
   'limit_mb',
   'save_delay_ms',
   'usage_flush_interval_sec',
+  'on_demand_refresh_min_interval_sec',
+  'on_demand_refresh_max_tokens',
   'upload_concurrent',
   'upload_timeout',
   'download_concurrent',
@@ -32,7 +34,10 @@ const NUMERIC_FIELDS = new Set([
   'medium_min_bytes',
   'blocked_parallel_attempts',
   'concurrent',
-  'batch_size'
+  'batch_size',
+  'max_file_size_mb',
+  'max_files',
+  'request_slow_ms'
 ]);
 
 const LOCALE_MAP = {
@@ -150,7 +155,18 @@ const LOCALE_MAP = {
     "fail_threshold": { title: "失败阈值", desc: "单个 Token 连续失败多少次后被标记为不可用。" },
     "save_delay_ms": { title: "保存延迟", desc: "Token 变更合并写入的延迟（毫秒）。" },
     "usage_flush_interval_sec": { title: "用量落库间隔", desc: "用量类字段写入数据库的最小间隔（秒）。" },
-    "reload_interval_sec": { title: "同步间隔", desc: "多 worker 场景下 Token 状态刷新间隔（秒）。" }
+    "reload_interval_sec": { title: "同步间隔", desc: "多 worker 场景下 Token 状态刷新间隔（秒）。" },
+    "on_demand_refresh_enabled": { title: "按需刷新", desc: "当请求拿不到可用 Token 时，是否允许触发受限的按需刷新。" },
+    "on_demand_refresh_min_interval_sec": { title: "按需刷新最小间隔", desc: "请求侧按需刷新之间的最小间隔（秒），用于避免刷新风暴。" },
+    "on_demand_refresh_max_tokens": { title: "按需刷新最大数量", desc: "单次请求侧按需刷新最多检查多少个 cooling Token。" }
+  },
+
+  "log": {
+    "label": "日志配置",
+    "max_file_size_mb": { title: "单文件上限", desc: "单个日志文件大小上限（MB），超过后自动轮转；设置为 0 或负数表示不按大小轮转。" },
+    "max_files": { title: "保留文件数", desc: "最多保留多少个日志文件；设置为 0 或负数表示不限制数量。" },
+    "log_all_requests": { title: "记录全部请求", desc: "开启后记录所有请求；关闭时仅记录慢请求、异常请求和错误请求。" },
+    "request_slow_ms": { title: "慢请求阈值", desc: "请求耗时超过该值（毫秒）时会写入日志。" }
   },
 
 

app/api/v1/admin/config.py

@@ -5,8 +5,13 @@
 
 from app.core.auth import verify_app_key
 from app.core.config import config
-from app.core.storage import get_storage as resolve_storage, LocalStorage, RedisStorage, SQLStorage
-from app.core.logger import logger
+from app.core.logger import logger, reload_logging_from_config
+from app.core.storage import (
+    LocalStorage,
+    RedisStorage,
+    SQLStorage,
+    get_storage as resolve_storage,
+)
 
 router = APIRouter()
 
@@ -99,6 +104,10 @@ async def update_config(data: dict):
     """更新配置"""
     try:
         await config.update(_sanitize_proxy_config_payload(data))
+        reload_logging_from_config(
+            default_level=os.getenv("LOG_LEVEL", "INFO"),
+            json_console=False,
+        )
         return {"status": "success", "message": "配置已更新"}
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))

app/core/logger.py

@@ -2,20 +2,20 @@
 结构化 JSON 日志 - 极简格式
 """
 
-import sys
-import os
 import json
+import os
+import sys
 import traceback
 from pathlib import Path
-from loguru import logger
+from typing import Any
 
-# Provide logging.Logger compatibility for legacy calls
-if not hasattr(logger, "isEnabledFor"):
-    logger.isEnabledFor = lambda _level: True
+from loguru import logger
 
 # 日志目录
 DEFAULT_LOG_DIR = Path(__file__).parent.parent.parent / "logs"
 LOG_DIR = Path(os.getenv("LOG_DIR", str(DEFAULT_LOG_DIR)))
+DEFAULT_LOG_MAX_FILE_SIZE_MB = 100
+DEFAULT_LOG_MAX_FILES = 7
 _LOG_DIR_READY = False
 
 
@@ -70,7 +70,13 @@ def _format_json(record) -> str:
             )
         )
 
-    return json.dumps(log_entry, ensure_ascii=False)
+    return json.dumps(log_entry, ensure_ascii=False, default=str)
+
+
+# Provide logging.Logger compatibility for legacy calls
+if not hasattr(logger, "isEnabledFor"):
+    logger.isEnabledFor = lambda _level: True
+
 
 def _env_flag(name: str, default: bool) -> bool:
     raw = os.getenv(name)
@@ -79,65 +85,114 @@ def _env_flag(name: str, default: bool) -> bool:
     return raw.strip().lower() in ("1", "true", "yes", "on", "y")
 
 
-def _make_json_sink(output):
-    """创建 JSON sink"""
-
-    def sink(message):
-        json_str = _format_json(message.record)
-        print(json_str, file=output, flush=True)
-
-    return sink
+def _env_int(name: str, default: int) -> int:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    try:
+        return int(raw.strip())
+    except (TypeError, ValueError):
+        return default
 
 
-def _file_json_sink(message):
-    """写入日志文件"""
-    record = message.record
-    json_str = _format_json(record)
-    log_file = LOG_DIR / f"app_{record['time'].strftime('%Y-%m-%d')}.log"
-    with open(log_file, "a", encoding="utf-8") as f:
-        f.write(json_str + "\n")
+def _patch_json_record(record) -> None:
+    """为全局 Loguru 记录补充序列化后的 JSON 文本。"""
+    record["extra"]["_json_line"] = _format_json(record)
 
 
 def setup_logging(
     level: str = "DEBUG",
     json_console: bool = True,
     file_logging: bool = True,
+    file_rotation_size_mb: int | None = None,
+    file_retention_count: int | None = None,
 ):
     """设置日志配置"""
+    logger.configure(patcher=_patch_json_record)
     logger.remove()
     file_logging = _env_flag("LOG_FILE_ENABLED", file_logging)
+    rotation_size_mb = _env_int(
+        "LOG_MAX_FILE_SIZE_MB",
+        DEFAULT_LOG_MAX_FILE_SIZE_MB
+        if file_rotation_size_mb is None
+        else int(file_rotation_size_mb),
+    )
+    retention_count = _env_int(
+        "LOG_MAX_FILES",
+        DEFAULT_LOG_MAX_FILES
+        if file_retention_count is None
+        else int(file_retention_count),
+    )
 
     # 控制台输出
     if json_console:
         logger.add(
-            _make_json_sink(sys.stdout),
+            sys.stdout,
             level=level,
-            format="{message}",
+            format="{extra[_json_line]}",
             colorize=False,
+            backtrace=False,
+            diagnose=False,
         )
     else:
         logger.add(
             sys.stdout,
             level=level,
             format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{file.name}:{line}</cyan> - <level>{message}</level>",
             colorize=True,
+            backtrace=False,
+            diagnose=False,
         )
 
     # 文件输出
     if file_logging:
         if _prepare_log_dir():
+            file_kwargs: dict[str, Any] = {
+                "level": level,
+                "format": "{extra[_json_line]}",
+                "colorize": False,
+                "enqueue": True,
+                "encoding": "utf-8",
+                "backtrace": False,
+                "diagnose": False,
+            }
+            if rotation_size_mb > 0:
+                file_kwargs["rotation"] = rotation_size_mb * 1024 * 1024
+            if retention_count > 0:
+                file_kwargs["retention"] = retention_count
             logger.add(
-                _file_json_sink,
-                level=level,
-                format="{message}",
-                enqueue=True,
+                str(LOG_DIR / "app_{time:YYYY-MM-DD}.log"),
+                **file_kwargs,
             )
         else:
             logger.warning("File logging disabled: no writable log directory.")
 
     return logger
 
 
+def reload_logging_from_config(
+    default_level: str = "INFO",
+    json_console: bool = False,
+):
+    """根据运行时配置重新加载日志设置。"""
+    try:
+        from app.core.config import get_config
+
+        return setup_logging(
+            level=str(get_config("log.level", default_level)),
+            json_console=json_console,
+            file_logging=bool(get_config("log.file_enabled", True)),
+            file_rotation_size_mb=get_config(
+                "log.max_file_size_mb", DEFAULT_LOG_MAX_FILE_SIZE_MB
+            ),
+            file_retention_count=get_config("log.max_files", DEFAULT_LOG_MAX_FILES),
+        )
+    except Exception as exc:
+        configured = setup_logging(level=default_level, json_console=json_console)
+        configured.warning("Failed to reload logging config: {}", exc)
+        return configured
+
+
 def get_logger(trace_id: str = "", span_id: str = ""):
     """获取绑定了 trace 上下文的 logger"""
     bound = {}
@@ -148,4 +203,10 @@ def get_logger(trace_id: str = "", span_id: str = ""):
     return logger.bind(**bound) if bound else logger
 
 
-__all__ = ["logger", "setup_logging", "get_logger", "LOG_DIR"]
+__all__ = [
+    "logger",
+    "setup_logging",
+    "reload_logging_from_config",
+    "get_logger",
+    "LOG_DIR",
+]

app/core/response_middleware.py

@@ -10,6 +10,7 @@
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
 
+from app.core.config import get_config
 from app.core.logger import logger
 
 
@@ -19,6 +20,23 @@ class ResponseLoggerMiddleware(BaseHTTPMiddleware):
     Request Logging and Response Tracking Middleware
     """
 
+    @staticmethod
+    def _should_log_response(path: str, status_code: int, duration_ms: float) -> bool:
+        if path == "/health" and not bool(
+            get_config("log.log_health_requests", False)
+        ):
+            return False
+
+        if bool(get_config("log.log_all_requests", False)):
+            return True
+
+        try:
+            slow_ms = float(get_config("log.request_slow_ms", 3000))
+        except (TypeError, ValueError):
+            slow_ms = 3000.0
+
+        return status_code >= 400 or duration_ms >= slow_ms
+
     async def dispatch(self, request: Request, call_next):
         # 生成请求 ID
         trace_id = str(uuid.uuid4())
@@ -40,39 +58,36 @@ async def dispatch(self, request: Request, call_next):
         ):
             return await call_next(request)
 
-        # 记录请求信息
-        logger.info(
-            f"Request: {request.method} {request.url.path}",
-            extra={
-                "traceID": trace_id,
-                "method": request.method,
-                "path": request.url.path,
-            },
-        )
-
         try:
             response = await call_next(request)
 
             # 计算耗时
             duration = (time.time() - start_time) * 1000
 
-            # 记录响应信息
-            logger.info(
-                f"Response: {request.method} {request.url.path} - {response.status_code} ({duration:.2f}ms)",
-                extra={
-                    "traceID": trace_id,
-                    "method": request.method,
-                    "path": request.url.path,
-                    "status": response.status_code,
-                    "duration_ms": round(duration, 2),
-                },
-            )
+            if self._should_log_response(path, response.status_code, duration):
+                log_method = (
+                    logger.error
+                    if response.status_code >= 500
+                    else logger.warning
+                    if response.status_code >= 400
+                    else logger.info
+                )
+                log_method(
+                    f"Response: {request.method} {request.url.path} - {response.status_code} ({duration:.2f}ms)",
+                    extra={
+                        "traceID": trace_id,
+                        "method": request.method,
+                        "path": request.url.path,
+                        "status": response.status_code,
+                        "duration_ms": round(duration, 2),
+                    },
+                )
 
             return response
 
         except Exception as e:
             duration = (time.time() - start_time) * 1000
-            logger.error(
+            logger.opt(exception=e).error(
                 f"Response Error: {request.method} {request.url.path} - {str(e)} ({duration:.2f}ms)",
                 extra={
                     "traceID": trace_id,
@@ -82,4 +97,4 @@ async def dispatch(self, request: Request, call_next):
                     "error": str(e),
                 },
             )
-            raise e
+            raise

app/services/grok/batch_services/usage.py

@@ -55,7 +55,7 @@ async def get(self, token: str) -> Dict:
                     remaining = data.get("remainingQueries")
                     if remaining is not None:
                         data["remainingTokens"] = remaining
-                logger.info(
+                logger.debug(
                     f"Usage sync success: remaining={remaining}, token={token[:10]}..."
                 )
                 return data

app/services/grok/utils/retry.py

@@ -25,7 +25,7 @@ async def pick_token(
             break
 
     if not token and not tried:
-        result = await token_mgr.refresh_cooling_tokens()
+        result = await token_mgr.refresh_cooling_tokens_on_demand()
         if result.get("recovered", 0) > 0:
             for pool_name in ModelService.pool_candidates_for_model(model_id):
                 token = token_mgr.get_token(pool_name, prefer_tags=prefer_tags)