160 lines
4.7 KiB
Python
160 lines
4.7 KiB
Python
"""logwriter.py —— 行为日志异步文件写入器(JSONL,固定目录,每日一个文件)。
|
|
|
|
设计:请求/业务只入队(不落盘),后台单任务批量 append,避免日志拖慢接口;
|
|
文件按日期命名 audit-YYYY-MM-DD.log,跨天自动切文件。写失败仅记 stderr,绝不影响业务。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_writer: "LogWriter | None" = None # 全局单例
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
|
|
|
|
|
|
def _today() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
|
|
class LogWriter:
|
|
def __init__(self, log_dir: str, retention_days: int = 0, hub_id: str = "") -> None:
|
|
self.log_dir = Path(log_dir)
|
|
self.retention_days = retention_days
|
|
self.hub_id = hub_id
|
|
self._queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=20000)
|
|
self._task: asyncio.Task | None = None
|
|
self._date = ""
|
|
self._fh = None
|
|
|
|
def set_hub_id(self, hub_id: str) -> None:
|
|
self.hub_id = hub_id or ""
|
|
|
|
async def start(self) -> None:
|
|
self.log_dir.mkdir(parents=True, exist_ok=True)
|
|
self._task = asyncio.create_task(self._run())
|
|
|
|
async def stop(self) -> None:
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._task = None
|
|
self._close_fh()
|
|
|
|
def enqueue(self, entry: dict[str, Any]) -> None:
|
|
try:
|
|
self._queue.put_nowait(entry)
|
|
except asyncio.QueueFull:
|
|
logger.warning("行为日志队列已满,丢弃一条")
|
|
|
|
def _close_fh(self) -> None:
|
|
if self._fh is not None:
|
|
try:
|
|
self._fh.close()
|
|
except Exception:
|
|
pass
|
|
self._fh = None
|
|
|
|
def _open_for_date(self, date_str: str) -> None:
|
|
self._close_fh()
|
|
path = self.log_dir / f"audit-{date_str}.log"
|
|
self._fh = open(path, "a", encoding="utf-8")
|
|
self._date = date_str
|
|
self._cleanup()
|
|
|
|
async def _run(self) -> None:
|
|
try:
|
|
while True:
|
|
batch: list[dict[str, Any]] = []
|
|
try:
|
|
batch.append(await asyncio.wait_for(self._queue.get(), timeout=1.0))
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
while len(batch) < 200:
|
|
try:
|
|
batch.append(self._queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
await asyncio.to_thread(self._write_sync, batch)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
|
|
def _write_sync(self, entries: list[dict[str, Any]]) -> None:
|
|
today = _today()
|
|
if self._date != today:
|
|
self._open_for_date(today)
|
|
for e in entries:
|
|
self._fh.write(json.dumps(e, ensure_ascii=False) + "\n")
|
|
self._fh.flush()
|
|
|
|
def _cleanup(self) -> None:
|
|
if not self.retention_days:
|
|
return
|
|
try:
|
|
import time as _time
|
|
|
|
cutoff = _time.time() - self.retention_days * 86400
|
|
for p in self.log_dir.glob("audit-*.log"):
|
|
try:
|
|
if p.stat().st_mtime < cutoff:
|
|
p.unlink()
|
|
except OSError:
|
|
continue
|
|
except Exception:
|
|
logger.exception("日志清理失败")
|
|
|
|
|
|
def init_writer(log_dir: str, retention_days: int = 0, hub_id: str = "") -> None:
|
|
global _writer
|
|
_writer = LogWriter(log_dir, retention_days, hub_id=hub_id)
|
|
|
|
|
|
async def start_writer() -> None:
|
|
if _writer is not None:
|
|
await _writer.start()
|
|
|
|
|
|
async def stop_writer() -> None:
|
|
global _writer
|
|
if _writer is not None:
|
|
await _writer.stop()
|
|
_writer = None
|
|
|
|
|
|
def log_event(
|
|
event: str,
|
|
*,
|
|
user_id: str | None = None,
|
|
method: str | None = None,
|
|
path: str | None = None,
|
|
status: int | None = None,
|
|
latency_ms: float | None = None,
|
|
detail: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""通用业务埋点;writer 未初始化时静默 no-op。"""
|
|
if _writer is None:
|
|
return
|
|
entry = {
|
|
"ts": _now_iso(),
|
|
"hub_id": _writer.hub_id,
|
|
"user_id": user_id,
|
|
"event": event,
|
|
"method": method,
|
|
"path": path,
|
|
"status": status,
|
|
"latency_ms": latency_ms,
|
|
"detail": detail or {},
|
|
}
|
|
_writer.enqueue(entry)
|