qihuaiflow/backend/python/app/core/observability.py

"""
Qihu-AIFlow 统一可观测基座 (Phase 2.1 + Phase 1.I OTel 增强)
=============================================================
提供贯穿一次请求的全局链路 ID（trace_id）与结构化日志能力，
为后续重构（上下文引擎、媒体/LLM 成本归因）建立回归基线。

设计原则（Strangler Fig 增量）：
- 纯增量埋点，不改任何既有接口的返回结构。
- 向后兼容现有 \"logging.getLogger(__name__)\" 用法：trace_id 由 root logger 的
  TraceIdFilter 自动注入，调用方无需改动。
- 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。

Phase 1.I 增强：span() 在保持原有日志计时的同时，额外嵌套一个标准 OpenTelemetry
span（经 app.core.otel 桥接，OTLP 导出到 Grafana）。OTel 失败时自动降级为 no-op，
绝不改变既有日志行为与请求链路。

使用方式：
    from app.core.observability import set_trace_id, get_trace_id, span

    # 在中间件里为每个请求注入：
    set_trace_id(request_id)

    # 在关键调用处打计时 span（自动带 trace_id + OTel span）：
    async with span("chat.send_message", conv_id=conv_id):
        return await _send_message_core(...)
"""
import logging
import time
import uuid
import json
from contextvars import ContextVar

from app.core.otel import current_span_cm, record_span_metrics

# ---------------------------------------------------------------------------
# trace_id：贯穿一次请求（含其内所有子调用/协程）的全局链路 ID
# ---------------------------------------------------------------------------
_trace_id_var: ContextVar[str] = ContextVar("aiflow_trace_id", default="-")


def new_trace_id() -> str:
    """生成一个崭新的链路 ID（32 位 hex）。"""
    return uuid.uuid4().hex


def set_trace_id(tid: str | None = None) -> str:
    """设置当前上下文的 trace_id；不传则自动生成。返回最终使用的 ID。"""
    tid = tid or new_trace_id()
    _trace_id_var.set(tid)
    return tid


def get_trace_id() -> str:
    """读取当前上下文的 trace_id；未设置时返回占位 '-'。"""
    return _trace_id_var.get()


class TraceIdFilter(logging.Filter):
    """为每一条日志记录注入 trace_id 属性（不存在则写为 '-'）。

    挂载到 root logger 后，所有 \"getLogger(__name__)\" 的子 logger 都会继承，
    从而让全平台既有日志自动带上链路 ID，零侵入。
    """

    def filter(self, record: logging.LogRecord) -> bool:
        if not hasattr(record, "trace_id"):
            try:
                record.trace_id = get_trace_id()
            except Exception:
                record.trace_id = "-"
        return True


# 标准 logging 字段，JSON 序列化时需排除，避免与自定义 extra 冲突
_STD_ATTRS = set(logging.makeLogRecord({}).__dict__.keys()) | {
    "message", "asctime", "msg", "trace_id", "exc_info", "stack_info", "args",
}


class JsonFormatter(logging.Formatter):
    """结构化 JSON 日志格式（通过环境变量 AIFLOW_JSON_LOGS=1 启用）。

    默认不启用，避免破坏现有日志采集/人工查阅习惯。
    """

    def format(self, record: logging.LogRecord) -> str:
        data = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "logger": record.name,
            "trace_id": getattr(record, "trace_id", "-"),
            "msg": record.getMessage(),
        }
        for k, v in record.__dict__.items():
            if k not in _STD_ATTRS:
                try:
                    json.dumps(v, ensure_ascii=False)
                    data[k] = v
                except (TypeError, ValueError):
                    data[k] = str(v)
        if record.exc_info:
            data["exc"] = self.formatException(record.exc_info)
        return json.dumps(data, ensure_ascii=False)


def get_logger(name: str | None = None) -> logging.Logger:
    """获取 logger（向后兼容现有用法）。trace_id 由 root filter 自动注入。"""
    return logging.getLogger(name)


# ---------------------------------------------------------------------------
# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段 + OTel 嵌套）
# ---------------------------------------------------------------------------
class _Span:
    """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态，并结束嵌套的 OTel span。"""

    def __init__(self, name: str, **fields):
        self.name = name
        self.fields = fields
        self._logger = logging.getLogger("aiflow.span")
        self._start = None
        self.elapsed_ms = 0.0
        self._otel_cm = None  # 嵌套的 OTel span 上下文管理器

    def _enter(self):
        self._start = time.perf_counter()
        self._logger.info(
            f"SPAN_START {self.name}",
            extra={"span": self.name, "event": "start", "trace_id": get_trace_id(), **self.fields},
        )
        # OTel 嵌套 span（失败安全：current_span_cm 返回 None 时 no-op）
        self._otel_cm = current_span_cm(self.name, self.fields)
        if self._otel_cm is not None:
            self._otel_cm.__enter__()
        return self

    def _exit(self, exc_type, exc, tb):
        self.elapsed_ms = (time.perf_counter() - self._start) * 1000
        status = "error" if exc_type else "ok"
        self._logger.info(
            f"SPAN_END {self.name} {self.elapsed_ms:.1f}ms",
            extra={
                "span": self.name,
                "event": "end",
                "elapsed_ms": round(self.elapsed_ms, 1),
                "status": status,
                "trace_id": get_trace_id(),
                **self.fields,
            },
        )
        # 结束嵌套的 OTel span（失败安全）
        if self._otel_cm is not None:
            try:
                from opentelemetry import trace as _ot
                if exc_type is not None:
                    self._otel_cm.set_status(
                        _ot.trace.Status(_ot.trace.StatusCode.ERROR, str(exc_type.__name__))
                    )
                else:
                    self._otel_cm.set_status(_ot.trace.Status(_ot.trace.StatusCode.OK))
                self._otel_cm.__exit__(exc_type, exc, tb)
            except Exception:
                pass
        # 记录 span 指标（耗时直方图 + 计数），供 Grafana 出图（失败安全）
        try:
            record_span_metrics(self.name, self.elapsed_ms, status)
        except Exception:
            pass
        return False  # 不吞掉异常

    # 同步协议
    def __enter__(self):
        return self._enter()

    def __exit__(self, exc_type, exc, tb):
        return self._exit(exc_type, exc, tb)

    # 异步协议
    async def __aenter__(self):
        return self._enter()

    async def __aexit__(self, exc_type, exc, tb):
        return self._exit(exc_type, exc, tb)


def span(name: str, **fields):
    """创建一个计时 span。可同步或异步 with。

    async with span("chat.send_message", conv_id=cid):
        ...
    """
    return _Span(name, **fields)


__all__ = [
    "new_trace_id",
    "set_trace_id",
    "get_trace_id",
    "TraceIdFilter",
    "JsonFormatter",
    "get_logger",
    "span",
]