150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
"""任务领取到商品页的脱敏分阶段计时。
|
|
|
|
日志字段使用白名单，只包含任务编号、稳定操作名、毫秒耗时和结果分类。
|
|
不要把 URL、控件树、异常正文或业务数据传入本模块。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from contextlib import contextmanager
|
|
from contextvars import ContextVar
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
import time
|
|
from typing import Callable, Iterator, Optional
|
|
|
|
from .db import data_dir
|
|
|
|
|
|
PERFORMANCE_LOG_NAME = "task_performance.jsonl"
|
|
_LOGGER_NAME = "cmautobuy.performance"
|
|
_ACTIVE_TRACE: ContextVar[Optional["TaskPerformanceTrace"]] = ContextVar(
|
|
"cmautobuy_task_performance_trace", default=None
|
|
)
|
|
|
|
|
|
def _default_sink(record: dict[str, object]) -> None:
|
|
logging.getLogger(_LOGGER_NAME).info(
|
|
json.dumps(record, ensure_ascii=False, separators=(",", ":"))
|
|
)
|
|
|
|
|
|
def configure_performance_logging(
|
|
directory: Optional[Path] = None,
|
|
) -> Path:
|
|
"""把性能事件写入可轮转 JSONL 文件，重复调用不会增加处理器。"""
|
|
|
|
log_directory = directory or data_dir() / "logs"
|
|
log_directory.mkdir(parents=True, exist_ok=True)
|
|
log_path = log_directory / PERFORMANCE_LOG_NAME
|
|
logger = logging.getLogger(_LOGGER_NAME)
|
|
if not any(
|
|
getattr(handler, "baseFilename", None) == str(log_path.resolve())
|
|
for handler in logger.handlers
|
|
):
|
|
handler = RotatingFileHandler(
|
|
log_path,
|
|
maxBytes=2 * 1024 * 1024,
|
|
backupCount=3,
|
|
encoding="utf-8",
|
|
)
|
|
handler.setFormatter(logging.Formatter("%(message)s"))
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.INFO)
|
|
logger.propagate = False
|
|
return log_path
|
|
|
|
|
|
class TaskPerformanceTrace:
|
|
"""收集一条任务的阶段耗时，并在绑定任务编号后输出记录。"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
monotonic: Callable[[], float] = time.monotonic,
|
|
sink: Callable[[dict[str, object]], None] = _default_sink,
|
|
) -> None:
|
|
self._monotonic = monotonic
|
|
self._sink = sink
|
|
self._started_at = monotonic()
|
|
self._task_id = ""
|
|
self._pending: list[tuple[str, int, str]] = []
|
|
|
|
@property
|
|
def task_id(self) -> str:
|
|
return self._task_id
|
|
|
|
def bind_task(self, task_id: str) -> None:
|
|
"""绑定稳定任务编号，并输出绑定前缓存的领取阶段记录。"""
|
|
|
|
value = str(task_id or "").strip()
|
|
if not value:
|
|
return
|
|
if self._task_id and self._task_id != value:
|
|
raise RuntimeError("性能计时不能跨任务复用")
|
|
self._task_id = value
|
|
pending, self._pending = self._pending, []
|
|
for operation, duration_ms, result in pending:
|
|
self._emit(operation, duration_ms, result)
|
|
|
|
@contextmanager
|
|
def stage(self, operation: str) -> Iterator[None]:
|
|
"""记录一个阶段；异常时只写 failed，不记录异常正文。"""
|
|
|
|
started_at = self._monotonic()
|
|
try:
|
|
yield
|
|
except BaseException:
|
|
self.record(operation, self._monotonic() - started_at, "failed")
|
|
raise
|
|
else:
|
|
self.record(operation, self._monotonic() - started_at, "succeeded")
|
|
|
|
def record(self, operation: str, seconds: float, result: str) -> None:
|
|
"""记录白名单阶段数据，负耗时会安全归零。"""
|
|
|
|
checked_operation = str(operation or "").strip()
|
|
checked_result = str(result or "").strip()
|
|
if not checked_operation or not checked_result:
|
|
raise ValueError("性能操作名和结果分类不能为空")
|
|
duration_ms = max(0, round(float(seconds) * 1000))
|
|
if self._task_id:
|
|
self._emit(checked_operation, duration_ms, checked_result)
|
|
else:
|
|
self._pending.append(
|
|
(checked_operation, duration_ms, checked_result)
|
|
)
|
|
|
|
def checkpoint(self, operation: str, result: str = "succeeded") -> None:
|
|
"""记录从本轮调度开始到当前时刻的端到端耗时。"""
|
|
|
|
self.record(operation, self._monotonic() - self._started_at, result)
|
|
|
|
@contextmanager
|
|
def activate(self) -> Iterator["TaskPerformanceTrace"]:
|
|
"""让同一工作线程内的设备和页面适配层取得本计时对象。"""
|
|
|
|
token = _ACTIVE_TRACE.set(self)
|
|
try:
|
|
yield self
|
|
finally:
|
|
_ACTIVE_TRACE.reset(token)
|
|
|
|
def _emit(self, operation: str, duration_ms: int, result: str) -> None:
|
|
self._sink(
|
|
{
|
|
"task_id": self._task_id,
|
|
"operation": operation,
|
|
"duration_ms": duration_ms,
|
|
"result": result,
|
|
}
|
|
)
|
|
|
|
|
|
def current_performance_trace() -> Optional[TaskPerformanceTrace]:
|
|
"""返回当前工作线程的任务计时对象；普通调用返回 None。"""
|
|
|
|
return _ACTIVE_TRACE.get()
|