407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""请求日志中间件：为每个请求记录时间、路径、耗时等，并提供统计查询。
|
||
|
||
- 内存环形缓冲区（最近 N 条）
|
||
- JSON Lines 文件持久化（自动轮转）
|
||
- ASGI 中间件透明捕获请求/响应
|
||
"""
|
||
|
||
import datetime
|
||
import json
|
||
import logging
|
||
import time
|
||
from collections import deque
|
||
from dataclasses import asdict, dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
logger = logging.getLogger("gateway.logging_middleware")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 数据结构
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class RequestLogEntry:
|
||
"""单条请求日志。"""
|
||
timestamp: str # ISO-8601
|
||
method: str
|
||
path: str
|
||
status_code: int
|
||
duration_ms: float
|
||
client_ip: str
|
||
worker: str = "" # 处理请求的 worker URL（空串表示网关本地处理）
|
||
response_code: Optional[int] = None # 响应 JSON 中的 code 字段
|
||
request_id: Optional[str] = None # 响应 JSON 中的 request_id
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 环形缓冲区
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class RingBuffer:
|
||
"""固定大小的环形缓冲区，线程安全。"""
|
||
|
||
def __init__(self, maxlen: int = 2000):
|
||
self._deque: deque = deque(maxlen=maxlen)
|
||
|
||
def append(self, entry: RequestLogEntry) -> None:
|
||
self._deque.append(entry)
|
||
|
||
def snapshot(self) -> List[RequestLogEntry]:
|
||
"""返回当前缓冲区副本（最新在前）。"""
|
||
return list(reversed(self._deque))
|
||
|
||
def __len__(self) -> int:
|
||
return len(self._deque)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# JSON Lines 文件写入（含轮转）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class LogFileWriter:
|
||
"""追加写入 JSON Lines 日志文件，自动按行数 / 天数轮转。
|
||
|
||
轮转策略：保留 1 个备份 (.jsonl.1)，不保留更多历史。
|
||
"""
|
||
|
||
def __init__(self, filepath: str, max_lines: int = 10000, max_age_days: int = 7):
|
||
self.filepath = Path(filepath)
|
||
self.max_lines = max_lines
|
||
self.max_age_seconds = max_age_days * 86400
|
||
|
||
def write(self, entry: RequestLogEntry) -> None:
|
||
try:
|
||
self._maybe_rotate()
|
||
self.filepath.parent.mkdir(parents=True, exist_ok=True)
|
||
line = json.dumps(asdict(entry), ensure_ascii=False) + "\n"
|
||
with open(self.filepath, "a", encoding="utf-8") as f:
|
||
f.write(line)
|
||
except Exception:
|
||
logger.warning("写入请求日志失败", exc_info=True)
|
||
|
||
def _maybe_rotate(self) -> None:
|
||
if not self.filepath.exists():
|
||
return
|
||
|
||
# 按天数轮转
|
||
mtime = self.filepath.stat().st_mtime
|
||
if time.time() - mtime > self.max_age_seconds:
|
||
self._rotate()
|
||
return
|
||
|
||
# 按行数轮转
|
||
try:
|
||
with open(self.filepath, "r", encoding="utf-8") as f:
|
||
count = sum(1 for _ in f)
|
||
if count >= self.max_lines:
|
||
self._rotate()
|
||
except Exception:
|
||
pass # 读不到就算了，下次再说
|
||
|
||
def _rotate(self) -> None:
|
||
backup = self.filepath.with_suffix(".jsonl.1")
|
||
if backup.exists():
|
||
backup.unlink()
|
||
try:
|
||
self.filepath.rename(backup)
|
||
logger.info("请求日志已轮转: %s → %s", self.filepath.name, backup.name)
|
||
except Exception:
|
||
logger.warning("日志轮转失败", exc_info=True)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 模块级全局状态（由 init_logging 初始化）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_buffer: Optional[RingBuffer] = None
|
||
_writer: Optional[LogFileWriter] = None
|
||
|
||
|
||
def init_logging(cfg: dict) -> None:
|
||
"""初始化日志缓冲区与文件写入器。"""
|
||
global _buffer, _writer
|
||
|
||
log_cfg = cfg.get("request_log", {})
|
||
if not log_cfg.get("enabled", True):
|
||
logger.info("请求日志已禁用")
|
||
return
|
||
|
||
log_file = log_cfg.get("log_file", "gateway/request_log.jsonl")
|
||
log_path = Path(log_file)
|
||
if not log_path.is_absolute():
|
||
log_path = Path(__file__).resolve().parent.parent / log_file
|
||
|
||
_buffer = RingBuffer(maxlen=log_cfg.get("buffer_size", 2000))
|
||
_writer = LogFileWriter(
|
||
filepath=str(log_path),
|
||
max_lines=log_cfg.get("max_file_lines", 10000),
|
||
max_age_days=log_cfg.get("max_file_age_days", 7),
|
||
)
|
||
|
||
# 从历史日志文件加载最近 N 条到缓冲区
|
||
buffer_size = log_cfg.get("buffer_size", 2000)
|
||
loaded = _load_from_logfile(str(log_path), buffer_size)
|
||
if loaded > 0:
|
||
logger.info("从日志文件恢复 %d 条历史记录", loaded)
|
||
|
||
logger.info("请求日志已启用 | 缓冲=%d | 文件=%s",
|
||
buffer_size, log_path)
|
||
|
||
|
||
def _load_from_logfile(filepath: str, max_entries: int) -> int:
|
||
"""从 JSON Lines 日志文件读取最近 max_entries 条到缓冲区。"""
|
||
try:
|
||
p = Path(filepath)
|
||
if not p.exists():
|
||
return 0
|
||
|
||
# 从文件末尾反向读取（高效处理大文件）
|
||
with open(p, "rb") as f:
|
||
# 估算：每条约 200 bytes，读最后 max_entries * 250 bytes 足够
|
||
chunk_size = max_entries * 250
|
||
f.seek(0, 2) # 文件末尾
|
||
file_size = f.tell()
|
||
read_size = min(chunk_size, file_size)
|
||
f.seek(max(0, file_size - read_size))
|
||
raw = f.read().decode("utf-8", errors="replace")
|
||
|
||
# 跳过可能不完整的第一行
|
||
lines = raw.split("\n")
|
||
if file_size > read_size:
|
||
# 第一行可能不完整，跳过
|
||
lines = lines[1:]
|
||
# 去掉末尾空行
|
||
lines = [l for l in lines if l.strip()]
|
||
|
||
# 只取最后 max_entries 条
|
||
lines = lines[-max_entries:]
|
||
|
||
count = 0
|
||
for line in lines:
|
||
try:
|
||
data = json.loads(line)
|
||
entry = RequestLogEntry(
|
||
timestamp=data.get("timestamp", ""),
|
||
method=data.get("method", ""),
|
||
path=data.get("path", ""),
|
||
status_code=data.get("status_code", 0),
|
||
duration_ms=data.get("duration_ms", 0.0),
|
||
client_ip=data.get("client_ip", ""),
|
||
worker=data.get("worker", ""),
|
||
response_code=data.get("response_code"),
|
||
request_id=data.get("request_id"),
|
||
)
|
||
_buffer.append(entry)
|
||
count += 1
|
||
except (json.JSONDecodeError, KeyError):
|
||
continue
|
||
|
||
return count
|
||
except Exception:
|
||
logger.warning("从日志文件恢复历史记录失败", exc_info=True)
|
||
return 0
|
||
|
||
|
||
def is_initialized() -> bool:
|
||
return _buffer is not None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ASGI 中间件
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _should_log(path: str) -> bool:
|
||
"""只记录 API 请求（/api/ 路径），跳过静态文件、健康检查等。"""
|
||
return path.startswith("/api/")
|
||
|
||
|
||
async def request_logging_middleware(request, call_next):
|
||
"""记录每个请求的耗时、状态码等信息。"""
|
||
|
||
# 未初始化或不需要记录的路径 → 直接放行
|
||
if _buffer is None or not _should_log(request.url.path):
|
||
return await call_next(request)
|
||
|
||
start = time.perf_counter()
|
||
|
||
# 获取客户端 IP（优先级：X-Forwarded-For > X-Real-IP > client.host）
|
||
client_ip = request.client.host if request.client else "unknown"
|
||
forwarded = request.headers.get("x-forwarded-for")
|
||
if forwarded:
|
||
client_ip = forwarded.split(",")[0].strip()
|
||
else:
|
||
real_ip = request.headers.get("x-real-ip")
|
||
if real_ip:
|
||
client_ip = real_ip.strip()
|
||
|
||
response = await call_next(request)
|
||
duration_ms = round((time.perf_counter() - start) * 1000, 2)
|
||
|
||
# 读取 worker URL（由 forward.py 在转发时写入 request.state）
|
||
worker_url = getattr(request.state, "worker_url", None) or ""
|
||
|
||
# 提取响应 body 并解析业务字段（仅 JSON 响应）
|
||
response_code = None
|
||
request_id = None
|
||
content_type = response.headers.get("content-type", "")
|
||
|
||
if "application/json" in content_type or "application/json" in (response.media_type or ""):
|
||
# 读取 body（兼容 body_iterator 和 body 两种属性）
|
||
body = getattr(response, "body", None)
|
||
if body is None:
|
||
body = b""
|
||
async for chunk in response.body_iterator:
|
||
body += chunk
|
||
try:
|
||
data = json.loads(body)
|
||
response_code = data.get("code")
|
||
request_id = data.get("request_id")
|
||
except (json.JSONDecodeError, UnicodeDecodeError):
|
||
pass
|
||
|
||
# 如果读取了 body_iterator，需要重建响应
|
||
if not hasattr(response, "body") or response.body is None:
|
||
from starlette.responses import Response as StarletteResponse
|
||
response = StarletteResponse(
|
||
content=body,
|
||
status_code=response.status_code,
|
||
headers=dict(response.headers),
|
||
media_type=response.media_type,
|
||
)
|
||
|
||
# 记录
|
||
now = datetime.datetime.utcnow()
|
||
entry = RequestLogEntry(
|
||
timestamp=now.strftime("%Y-%m-%dT%H:%M:%S.") +
|
||
f"{now.microsecond // 1000:03d}Z",
|
||
method=request.method,
|
||
path=request.url.path,
|
||
status_code=response.status_code,
|
||
duration_ms=duration_ms,
|
||
client_ip=client_ip,
|
||
worker=worker_url,
|
||
response_code=response_code,
|
||
request_id=request_id,
|
||
)
|
||
|
||
_buffer.append(entry)
|
||
if _writer is not None:
|
||
_writer.write(entry)
|
||
|
||
return response
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 统计查询
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def get_stats() -> Dict[str, Any]:
|
||
"""基于缓冲区数据计算统计摘要，返回给统计页面使用。"""
|
||
if _buffer is None:
|
||
return {
|
||
"summary": {"total": 0, "success_rate": 0, "avg_duration_ms": 0,
|
||
"min_duration_ms": 0, "max_duration_ms": 0},
|
||
"endpoints": [],
|
||
"workers": [],
|
||
"recent": [],
|
||
"last_updated": datetime.datetime.utcnow().isoformat() + "Z",
|
||
}
|
||
|
||
snapshot = _buffer.snapshot()
|
||
total = len(snapshot)
|
||
|
||
if total == 0:
|
||
return {
|
||
"summary": {"total": 0, "success_rate": 0, "avg_duration_ms": 0,
|
||
"min_duration_ms": 0, "max_duration_ms": 0},
|
||
"endpoints": [],
|
||
"workers": [],
|
||
"recent": [],
|
||
"last_updated": datetime.datetime.utcnow().isoformat() + "Z",
|
||
}
|
||
|
||
# 汇总指标
|
||
durations = [e.duration_ms for e in snapshot]
|
||
ok_count = sum(1 for e in snapshot if e.response_code == 0)
|
||
|
||
# 按路径聚合
|
||
by_path: Dict[str, dict] = {}
|
||
for e in snapshot:
|
||
path = e.path
|
||
if path not in by_path:
|
||
by_path[path] = {"count": 0, "total_duration": 0.0, "ok": 0}
|
||
by_path[path]["count"] += 1
|
||
by_path[path]["total_duration"] += e.duration_ms
|
||
if e.response_code == 0:
|
||
by_path[path]["ok"] += 1
|
||
|
||
endpoints = sorted(
|
||
({
|
||
"path": path,
|
||
"count": v["count"],
|
||
"avg_duration_ms": round(v["total_duration"] / v["count"], 1),
|
||
"max_duration_ms": round(
|
||
max(e.duration_ms for e in snapshot if e.path == path), 1),
|
||
"success_rate": round(v["ok"] / v["count"] * 100, 1),
|
||
} for path, v in by_path.items()),
|
||
key=lambda x: -x["count"],
|
||
)
|
||
|
||
# 最近 100 条（最新在前）
|
||
recent_100 = snapshot[:100]
|
||
recent = [
|
||
{
|
||
"timestamp": e.timestamp,
|
||
"method": e.method,
|
||
"path": e.path,
|
||
"status_code": e.status_code,
|
||
"duration_ms": e.duration_ms,
|
||
"client_ip": e.client_ip,
|
||
"worker": e.worker,
|
||
"response_code": e.response_code,
|
||
"request_id": e.request_id,
|
||
}
|
||
for e in recent_100
|
||
]
|
||
|
||
# 按 worker 聚合
|
||
by_worker: Dict[str, dict] = {}
|
||
for e in snapshot:
|
||
w = e.worker or "(网关本地)"
|
||
if w not in by_worker:
|
||
by_worker[w] = {"count": 0, "total_duration": 0.0, "ok": 0}
|
||
by_worker[w]["count"] += 1
|
||
by_worker[w]["total_duration"] += e.duration_ms
|
||
if e.response_code == 0:
|
||
by_worker[w]["ok"] += 1
|
||
|
||
workers = sorted(
|
||
({
|
||
"worker": w,
|
||
"count": v["count"],
|
||
"avg_duration_ms": round(v["total_duration"] / v["count"], 1),
|
||
"success_rate": round(v["ok"] / v["count"] * 100, 1) if v["count"] else 0,
|
||
} for w, v in by_worker.items()),
|
||
key=lambda x: -x["count"],
|
||
)
|
||
|
||
return {
|
||
"summary": {
|
||
"total": total,
|
||
"success_rate": round(ok_count / total * 100, 1),
|
||
"avg_duration_ms": round(sum(durations) / len(durations), 1),
|
||
"min_duration_ms": round(min(durations), 1),
|
||
"max_duration_ms": round(max(durations), 1),
|
||
},
|
||
"endpoints": endpoints,
|
||
"workers": workers,
|
||
"recent": recent,
|
||
"last_updated": datetime.datetime.utcnow().isoformat() + "Z",
|
||
}
|