添加统计请求的功能
This commit is contained in:
Ubuntu
@@ -0,0 +1,406 @@
|
||||
"""请求日志中间件：为每个请求记录时间、路径、耗时等，并提供统计查询。
|
||||
|
||||
- 内存环形缓冲区（最近 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",
|
||||
}
|
||||
Reference in New Issue
Block a user