167 lines
5.8 KiB
Python
167 lines
5.8 KiB
Python
"""
|
||
access 日志采样 + 异步日志队列（QueueHandler + QueueListener）测试
|
||
|
||
验证:
|
||
1. _should_log_access：错误（>=400）全量；成功按采样率（确定性计数）；1.0 全量
|
||
2. QueueHandler + QueueListener：logger 只挂队列 handler，I/O 由后台监听线程落盘
|
||
3. 队列满不抛异常（丢弃而非阻塞/打断业务）
|
||
4. 端到端：真实 app 请求 -> access 日志异步落盘（含采样命中）
|
||
|
||
运行: uv run python test_logging_async.py
|
||
"""
|
||
import importlib.util
|
||
import itertools
|
||
import queue
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "gateway" / "src"))
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "utils" / "worker" / "src"))
|
||
|
||
_tmp = tempfile.TemporaryDirectory()
|
||
|
||
# 预加载真实 logger 模块（不触发 chongming 包链导入），把日志目录重定向到
|
||
# 临时目录，之后 import chongming.app 时 get_logger 即用新路径
|
||
_logger_path = (
|
||
Path(__file__).resolve().parent.parent / "gateway" / "src" / "chongming" / "app" / "core" / "logger.py"
|
||
)
|
||
_logger_spec = importlib.util.spec_from_file_location("chongming.app.core.logger", _logger_path)
|
||
logger_mod = importlib.util.module_from_spec(_logger_spec)
|
||
sys.modules["chongming.app.core.logger"] = logger_mod
|
||
_logger_spec.loader.exec_module(logger_mod) # type: ignore[union-attr]
|
||
logger_mod.log_config["file"] = _tmp.name
|
||
|
||
import chongming.app as app_mod
|
||
|
||
failures = []
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
status = "PASS" if cond else "FAIL"
|
||
print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
|
||
if not cond:
|
||
failures.append(name)
|
||
|
||
|
||
def _wait_for(pred, timeout=3.0):
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
if pred():
|
||
return True
|
||
time.sleep(0.05)
|
||
return False
|
||
|
||
|
||
# ---- 1. 采样逻辑 ----
|
||
def test_sampling():
|
||
app_mod._access_log_counter = itertools.count()
|
||
check(
|
||
"错误 4xx/5xx 全量",
|
||
app_mod._should_log_access(404)
|
||
and app_mod._should_log_access(500)
|
||
and app_mod._should_log_access(503),
|
||
)
|
||
app_mod._access_log_counter = itertools.count()
|
||
check("rate>=1.0 全量", app_mod._should_log_access(200, 1.0) and app_mod._should_log_access(200, 1.0))
|
||
app_mod._access_log_counter = itertools.count()
|
||
logged = sum(1 for _ in range(10) if app_mod._should_log_access(200, 0.1))
|
||
check("0.1 采样率 10 个成功请求记 1 条", logged == 1, f"logged={logged}")
|
||
|
||
|
||
# ---- 2. 异步队列：logger 只挂队列 handler，监听线程落盘 ----
|
||
def test_async_queue_write():
|
||
from logging.handlers import QueueHandler
|
||
|
||
lg = logger_mod.get_logger("test_async")
|
||
check("logger 只挂队列 handler", len(lg.handlers) == 1 and isinstance(lg.handlers[0], QueueHandler))
|
||
check("监听线程已启动", logger_mod._listener is not None and logger_mod._listener._thread.is_alive())
|
||
|
||
for i in range(5):
|
||
lg.info("async log %d", i) # % 参数化：格式化在监听线程执行
|
||
ok = _wait_for(
|
||
lambda: any(
|
||
"async log 4" in f.read_text(errors="ignore")
|
||
for f in Path(_tmp.name).glob("test_async-*.log")
|
||
)
|
||
)
|
||
check("日志异步落盘（监听线程写入）", ok)
|
||
|
||
|
||
# ---- 3. 队列满：丢弃而非抛异常 ----
|
||
def test_queue_full_drops():
|
||
old = logger_mod._log_queue
|
||
logger_mod._log_queue = queue.Queue(maxsize=2) # 换小队列
|
||
try:
|
||
lg = logger_mod.get_logger("test_full")
|
||
try:
|
||
for i in range(20):
|
||
lg.info("flood %d", i)
|
||
check("队列满不抛异常（丢弃新日志）", True)
|
||
except Exception as e:
|
||
check("队列满不抛异常（丢弃新日志）", False, str(e))
|
||
finally:
|
||
logger_mod._log_queue = old
|
||
|
||
|
||
# ---- 4. 按名隔离：不同 log_name 的记录不串文件 ----
|
||
def test_name_isolation():
|
||
logger_mod.get_logger("test_b")
|
||
lg_a = logger_mod.get_logger("test_async")
|
||
lg_b = logger_mod.get_logger("test_b")
|
||
lg_a.info("隔离测试 A 专属 %d", 1)
|
||
lg_b.info("隔离测试 B 专属 %d", 2)
|
||
ok = _wait_for(
|
||
lambda: any(
|
||
"隔离测试 B 专属 2" in f.read_text(errors="ignore")
|
||
for f in Path(_tmp.name).glob("test_b-*.log")
|
||
)
|
||
)
|
||
check("test_b 记录写入 test_b 文件", ok)
|
||
a_text = "".join(f.read_text(errors="ignore") for f in Path(_tmp.name).glob("test_async-*.log"))
|
||
b_text = "".join(f.read_text(errors="ignore") for f in Path(_tmp.name).glob("test_b-*.log"))
|
||
check("test_async 记录不串入 test_b 文件", "隔离测试 A 专属" not in b_text)
|
||
check("test_b 记录不串入 test_async 文件", "隔离测试 B 专属" not in a_text)
|
||
|
||
|
||
# ---- 5. 端到端：中间件采样 + 异步落盘 ----
|
||
def test_e2e_middleware():
|
||
import asyncio
|
||
import httpx
|
||
|
||
app_mod._access_log_counter = itertools.count() # 重置采样计数
|
||
|
||
async def main():
|
||
transport = httpx.ASGITransport(app=app_mod.app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
r = await ac.get("/")
|
||
return r.status_code
|
||
|
||
code = asyncio.run(main())
|
||
check("请求返回 200", code == 200, str(code))
|
||
ok = _wait_for(
|
||
lambda: any(
|
||
"GET / status=200" in f.read_text(errors="ignore")
|
||
for f in Path(_tmp.name).glob("app-*.log")
|
||
)
|
||
)
|
||
check("access 日志异步落盘（采样命中）", ok)
|
||
|
||
|
||
test_sampling()
|
||
test_async_queue_write()
|
||
test_queue_full_drops()
|
||
test_name_isolation()
|
||
test_e2e_middleware()
|
||
|
||
# 关闭监听线程并释放文件句柄，再清理临时目录
|
||
logger_mod.shutdown_logging()
|
||
_tmp.cleanup()
|
||
|
||
print()
|
||
if failures:
|
||
print(f"共 {len(failures)} 项失败: {failures}")
|
||
sys.exit(1)
|
||
print("全部通过")
|