feat(applog): 客户端日志专用落盘 writer(白名单+data 兜底, propagate=False)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
guke
@@ -0,0 +1,145 @@
|
||||
"""客户端运行日志专用落盘 writer(独立于服务端 app-server.log)。
|
||||
|
||||
- 独占 logger "client" + 自己的 RotatingFileHandler,propagate=False → 不污染 app-server.log。
|
||||
- 每条按「白名单键(client_ts/level/trace_id/tag/msg)提顶层 + 其余并入 data」封装,再
|
||||
json.dumps 成一行写出(钉死 SLS 索引列;见 spec §5)。formatter 用 %(message)s——行本身
|
||||
已是 JSON,不能再过 JsonFormatter 二次编码。
|
||||
- 滚动 20MB×10(env 可调),与服务日志同机制。
|
||||
⚠️ 依赖 --workers 1:RotatingFileHandler 多进程并发 doRollover 会损坏/丢日志;扩 worker
|
||||
前换 QueueHandler→单写入者 / 外部 logrotate(copytruncate) / 写 stdout 交 journald。
|
||||
|
||||
服务端补的字段(time/source/service/client_ip/device_id/...)是「事实」,与客户端自述分开。
|
||||
`time` 用服务端接收时间作 SLS 主时间(客户端时钟不可信),client_ts 另存为可查字段。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from datetime import datetime
|
||||
from logging.handlers import RotatingFileHandler
|
||||
from pathlib import Path
|
||||
|
||||
# 仅这些客户端键提到输出行顶层;其余(含客户端自带 data)一律并入 data,防 SLS 索引列爆炸
|
||||
_TOP_LEVEL_KEYS = ("client_ts", "level", "trace_id", "tag", "msg")
|
||||
|
||||
_logger: logging.Logger | None = None
|
||||
|
||||
|
||||
def _max_msg_bytes() -> int:
|
||||
return int(os.getenv("APPLOG_MAX_MSG_BYTES", "8192"))
|
||||
|
||||
|
||||
def _build_logger() -> logging.Logger:
|
||||
lg = logging.getLogger("client")
|
||||
lg.setLevel(logging.INFO)
|
||||
lg.propagate = False # 不冒泡到 root → 不写进 app-server.log
|
||||
log_file = os.getenv("CLIENT_LOG_FILE") or str(
|
||||
Path(os.getenv("LOG_DIR", "logs")) / "app-client.log"
|
||||
)
|
||||
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
|
||||
handler = RotatingFileHandler(
|
||||
log_file,
|
||||
maxBytes=int(os.getenv("CLIENT_LOG_MAX_BYTES", str(20 * 1024 * 1024))),
|
||||
backupCount=int(os.getenv("CLIENT_LOG_BACKUP_COUNT", "10")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
handler.setFormatter(logging.Formatter("%(message)s")) # 行已是 JSON,不再包装
|
||||
lg.handlers = [handler]
|
||||
return lg
|
||||
|
||||
|
||||
def get_logger() -> logging.Logger:
|
||||
global _logger
|
||||
if _logger is None:
|
||||
_logger = _build_logger()
|
||||
return _logger
|
||||
|
||||
|
||||
def reset_client_logger() -> None:
|
||||
"""测试用:关闭并丢弃当前 logger,使下次 get_logger 按当时 env 重建(切临时文件)。"""
|
||||
global _logger
|
||||
if _logger is not None:
|
||||
for h in list(_logger.handlers):
|
||||
h.close()
|
||||
_logger.handlers = []
|
||||
_logger = None
|
||||
|
||||
|
||||
def _truncate_msg(msg: str) -> tuple[str, bool]:
|
||||
raw = msg.encode("utf-8")
|
||||
limit = _max_msg_bytes()
|
||||
if len(raw) <= limit:
|
||||
return msg, False
|
||||
# 按字节截断后解码,忽略截断处半个多字节字符
|
||||
return raw[:limit].decode("utf-8", "ignore") + "…[truncated]", True
|
||||
|
||||
|
||||
def _build_line(
|
||||
record: dict, *, meta: dict, client_ip: str, service: str, now_iso: str
|
||||
) -> str:
|
||||
out: dict = {
|
||||
"time": now_iso,
|
||||
"source": "client",
|
||||
"service": service,
|
||||
"client_ip": client_ip,
|
||||
}
|
||||
# 批级公共字段(非空才带)
|
||||
for k in ("device_id", "user_id", "app_ver", "platform", "sent_at"):
|
||||
v = meta.get(k)
|
||||
if v is not None:
|
||||
out[k] = v
|
||||
# 白名单键提顶层
|
||||
if record.get("level") is not None:
|
||||
out["level"] = str(record["level"]).upper()
|
||||
if record.get("trace_id"):
|
||||
out["trace_id"] = record["trace_id"]
|
||||
if record.get("tag"):
|
||||
out["tag"] = record["tag"]
|
||||
if record.get("client_ts") is not None:
|
||||
out["client_ts"] = record["client_ts"]
|
||||
if record.get("msg") is not None:
|
||||
msg, truncated = _truncate_msg(str(record["msg"]))
|
||||
out["msg"] = msg
|
||||
if truncated:
|
||||
out["msg_truncated"] = True
|
||||
# 其余键(含客户端自带 data)并入 data
|
||||
data: dict = {}
|
||||
client_data = record.get("data")
|
||||
if isinstance(client_data, dict):
|
||||
data.update(client_data)
|
||||
for k, v in record.items():
|
||||
if k in _TOP_LEVEL_KEYS or k == "data":
|
||||
continue
|
||||
data[k] = v
|
||||
if data:
|
||||
out["data"] = data
|
||||
return json.dumps(out, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
def write_records(
|
||||
records: list[dict], *, meta: dict, client_ip: str
|
||||
) -> tuple[int, int]:
|
||||
"""把一批客户端日志逐条写入专用文件。返回 (received, dropped)。
|
||||
|
||||
尽力而为(fire-and-forget):logger 初始化或单条写入失败只跳过并计 dropped,
|
||||
不抛给上层——端点因此永不因写日志而 500。
|
||||
"""
|
||||
try:
|
||||
lg = get_logger()
|
||||
except Exception: # noqa: BLE001 — 初始化失败也不能让端点 500
|
||||
logging.getLogger("shagua.applog").exception("client log writer init failed")
|
||||
return 0, len(records)
|
||||
service = os.getenv("CLIENT_LOG_SERVICE_NAME", "app-client")
|
||||
now_iso = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
|
||||
received = dropped = 0
|
||||
for rec in records:
|
||||
try:
|
||||
line = _build_line(
|
||||
rec, meta=meta, client_ip=client_ip, service=service, now_iso=now_iso
|
||||
)
|
||||
lg.info(line)
|
||||
received += 1
|
||||
except Exception: # noqa: BLE001 — 坏条跳过,不影响其余
|
||||
dropped += 1
|
||||
return received, dropped
|
||||
@@ -0,0 +1,95 @@
|
||||
"""客户端运行日志上报:writer 单测 + 端点集成测试。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core import client_log
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client_log_file(tmp_path, monkeypatch):
|
||||
"""把客户端日志切到临时文件,并重置 writer 单例使其按当时 env 重建。"""
|
||||
p = tmp_path / "app-client.log"
|
||||
monkeypatch.setenv("CLIENT_LOG_FILE", str(p))
|
||||
client_log.reset_client_logger()
|
||||
yield p
|
||||
client_log.reset_client_logger()
|
||||
|
||||
|
||||
def _read_lines(p: Path) -> list[dict]:
|
||||
text = p.read_text(encoding="utf-8").strip()
|
||||
return [json.loads(ln) for ln in text.splitlines() if ln]
|
||||
|
||||
|
||||
# ---------------- writer 层 ----------------
|
||||
|
||||
def _meta(**kw) -> dict:
|
||||
base = {"device_id": "d-1", "user_id": None, "app_ver": None,
|
||||
"platform": None, "sent_at": None}
|
||||
base.update(kw)
|
||||
return base
|
||||
|
||||
|
||||
def test_writer_writes_one_line_per_record(client_log_file):
|
||||
recs = [
|
||||
{"client_ts": 1737000000000, "level": "info", "msg": "hello"},
|
||||
{"client_ts": 1737000000001, "level": "error", "msg": "boom", "tag": "net"},
|
||||
]
|
||||
received, dropped = client_log.write_records(
|
||||
recs, meta=_meta(user_id=42, app_ver="1.2.3", platform="android"),
|
||||
client_ip="1.2.3.4",
|
||||
)
|
||||
assert (received, dropped) == (2, 0)
|
||||
lines = _read_lines(client_log_file)
|
||||
assert len(lines) == 2
|
||||
assert lines[0]["source"] == "client"
|
||||
assert lines[0]["service"] == "app-client"
|
||||
assert lines[0]["device_id"] == "d-1"
|
||||
assert lines[0]["user_id"] == 42
|
||||
assert lines[0]["app_ver"] == "1.2.3"
|
||||
assert lines[0]["client_ip"] == "1.2.3.4"
|
||||
assert lines[0]["level"] == "INFO" # 归一化大写
|
||||
assert lines[0]["msg"] == "hello"
|
||||
assert lines[0]["client_ts"] == 1737000000000
|
||||
assert lines[1]["tag"] == "net"
|
||||
|
||||
|
||||
def test_writer_hoists_trace_id_to_top_level(client_log_file):
|
||||
client_log.write_records(
|
||||
[{"client_ts": 1, "level": "info", "msg": "x", "trace_id": "abc123"}],
|
||||
meta=_meta(), client_ip="",
|
||||
)
|
||||
assert _read_lines(client_log_file)[0]["trace_id"] == "abc123"
|
||||
|
||||
|
||||
def test_writer_sweeps_unknown_keys_into_data(client_log_file):
|
||||
client_log.write_records(
|
||||
[{"client_ts": 1, "level": "info", "msg": "x",
|
||||
"foo": 123, "data": {"bar": "baz"}}],
|
||||
meta=_meta(), client_ip="",
|
||||
)
|
||||
line = _read_lines(client_log_file)[0]
|
||||
assert "foo" not in line # 白名单外不进顶层
|
||||
assert line["data"]["foo"] == 123 # 兜底进 data
|
||||
assert line["data"]["bar"] == "baz" # 客户端自带 data 合并进来
|
||||
|
||||
|
||||
def test_writer_truncates_oversize_msg(client_log_file, monkeypatch):
|
||||
monkeypatch.setenv("APPLOG_MAX_MSG_BYTES", "10")
|
||||
received, dropped = client_log.write_records(
|
||||
[{"client_ts": 1, "level": "info", "msg": "x" * 100}],
|
||||
meta=_meta(), client_ip="",
|
||||
)
|
||||
assert (received, dropped) == (1, 0) # 截断而非丢弃
|
||||
line = _read_lines(client_log_file)[0]
|
||||
assert line["msg_truncated"] is True
|
||||
assert line["msg"].endswith("…[truncated]")
|
||||
|
||||
|
||||
def test_writer_logger_does_not_propagate(client_log_file):
|
||||
lg = client_log.get_logger()
|
||||
assert lg.name == "client"
|
||||
assert lg.propagate is False # 不冒泡到 root → 不写 app-server.log
|
||||
Reference in New Issue
Block a user