long_articles_agentic_supply/supply/infra/observability/logging.py

"""异步日志服务 —— 基于 Queue + 后台 worker 推送到阿里云 SLS

职责：
  - 接收结构化 LogRecord，序列化后批量推送
  - 提供 log_raw() 向下兼容旧的 dict 调用方式
  - Queue 满时静默丢弃，不阻塞业务
"""

import asyncio
import contextlib
import datetime
import json
import logging
import time
import traceback
from typing import Optional

from aliyun.log import LogClient, LogItem, PutLogsRequest

from supply.config.aliyun import AliyunLogConfig
from supply.infra.observability.log_schema import LogRecord

logger = logging.getLogger(__name__)


class LogService:
    """异步日志服务：将结构化日志放入 Queue，后台 worker 批量推送到 SLS

    使用方式：
        service = LogService(AliyunLogConfig())
        await service.start()

        # 结构化方式（推荐）
        record = LogRecord(
            level=LogLevel.INFO,
            category=LogCategory.TASK,
            event="task_started",
            trace_id="trace-123",
            task_name="content_generation",
        )
        await service.log(record)

        # 向后兼容：自由 dict
        await service.log_raw({"event": "legacy_call", "key": "val"})

        await service.stop()
    """

    def __init__(self, log_config: AliyunLogConfig):
        self.config = log_config
        self.client: Optional[LogClient] = None
        self.queue: Optional[asyncio.Queue] = None
        self._worker_task: Optional[asyncio.Task] = None
        self._running = False

    # ==================== 生命周期 ====================

    async def start(self):
        if self._running:
            return
        self.client = LogClient(
            self.config.endpoint,
            self.config.access_key_id,
            self.config.access_key_secret,
        )
        self.queue = asyncio.Queue(maxsize=10000)
        self._running = True
        self._worker_task = asyncio.create_task(self._worker())
        logger.info("LogService 已启动 → %s/%s", self.config.project, self.config.logstore)

    async def stop(self):
        if not self._running:
            return
        self._running = False
        if self._worker_task:
            self._worker_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._worker_task
        self._worker_task = None
        self.queue = None
        self.client = None
        logger.info("LogService 已停止")

    # ==================== 对外接口 ====================

    async def log(self, record: LogRecord) -> None:
        """推送结构化日志（推荐方式，非阻塞）"""
        if not self._running or self.queue is None:
            return
        try:
            self.queue.put_nowait(record)
        except asyncio.QueueFull:
            pass

    async def log_raw(self, contents: dict) -> None:
        """向后兼容：接受自由 dict，内部转为 LogRecord 推送

        所有值会被 str() 化以匹配 SLS API 要求。
        """
        if not self._running or self.queue is None:
            return
        try:
            # 从 dict 提取已知字段，其余塞入 extra
            level_str = str(contents.pop("level", "INFO")).upper()
            category_str = str(contents.pop("category", "system")).lower()
            try:
                from supply.infra.observability.log_schema import LogLevel, LogCategory

                level = LogLevel(level_str) if level_str in LogLevel._value2member_map_ else LogLevel.INFO
                category = LogCategory(category_str) if category_str in LogCategory._value2member_map_ else LogCategory.SYSTEM
            except (KeyError, ValueError):
                level = LogLevel.INFO
                category = LogCategory.SYSTEM

            record = LogRecord(
                level=level,
                category=category,
                event=str(contents.pop("event", "")),
                message=str(contents.pop("message", "")),
                trace_id=str(contents.pop("trace_id", "")),
                task_name=str(contents.pop("task_name", "")),
                agent_name=str(contents.pop("agent_name", "")),
                tool_name=str(contents.pop("tool_name", "")),
                duration_ms=int(contents.pop("duration_ms", 0)),
                status=str(contents.pop("status", "success")),
                error_type=str(contents.pop("error_type", "")),
                error_message=str(contents.pop("error_message", "")),
                extra=contents,  # 剩余字段
            )
            self.queue.put_nowait(record)
        except asyncio.QueueFull:
            pass

    # ==================== 后台 Worker ====================

    async def _worker(self):
        try:
            while self._running:
                record = await self.queue.get()
                try:
                    await asyncio.to_thread(self._put_log, record)
                except Exception as e:
                    print(f"[Log Error] {e}")
                    print(traceback.format_exc())
        except asyncio.CancelledError:
            pass

    def _put_log(self, record: LogRecord):
        """同步推送到 SLS"""
        timestamp = int(time.time())

        # LogRecord → SLS LogItem
        items = record.to_sls_items()
        # 追加内置字段
        items.append(("datetime", datetime.datetime.now().isoformat()))

        log_item = LogItem(timestamp=timestamp, contents=items)
        req = PutLogsRequest(
            self.config.project,
            self.config.logstore,
            topic=record.category.value,     # topic 用 category，方便消费端路由
            source="",
            logitems=[log_item],
        )
        self.client.put_logs(req)