long_articles_agentic_supply/supply/infra/observability/log_schema.py

"""结构化日志模型 + SLS 索引配置

设计目标：
  1. 统一日志字段，消除当前 free-form dict 导致的查询困难
  2. 按查询模式设计 SLS 键值索引，避免全表扫描
  3. 支持 trace_id 串联完整调用链


====== SLS 索引配置说明 ======

全文索引（分词查询）:
  字段              分词符              包含中文
  ───────────────────────────────────────────
  message            ,;=|'"\t\r\n         ✓
  error_message      ,;=|'"\t\r\n         ✓

键值索引（精确匹配 / 范围 / 聚合）:
  字段              类型        别名              说明
  ────────────────────────────────────────────────────
  level             text        log_level         日志级别，精确匹配 + 统计
  category          text        log_category      日志分类，精确匹配
  event             text                          事件名，精确匹配 + 前缀查询
  trace_id          text                          分布式追踪 ID，精确匹配
  task_name         text                          任务名，精确匹配 + 前缀查询
  agent_name        text                          Agent 名，精确匹配
  tool_name         text                          Tool 名，精确匹配
  duration_ms       long                          耗时（毫秒），范围查询 + 统计
  status            text                          执行状态，精确匹配
  error_type        text                          异常类型，精确匹配 + 聚合
  extra             json                          扩展字段，子字段可索引

统计优化:
  duration_ms 开启统计（min / max / avg / sum / distinct_count）
  level 开启统计（distinct_count）

控制台创建 SQL 建索引（参考，实际通过控制台或 API 创建）:
  -- 全文索引
  {"Line": {"Token": [",;=|'\"\t\r\n"]}, "Keys": {"message": {"Token": [",;=|'\"\t\r\n"], "CaseSensitive": false, "Chn": true}, "error_message": {"Token": [",;=|'\"\t\r\n"], "CaseSensitive": false, "Chn": true}}}

  -- 键值索引
  Key=level          Type=text
  Key=category       Type=text
  Key=event          Type=text
  Key=trace_id       Type=text
  Key=task_name      Type=text
  Key=agent_name     Type=text
  Key=tool_name      Type=text
  Key=duration_ms    Type=long  Stats=true
  Key=status         Type=text
  Key=error_type     Type=text
  Key=extra          Type=json


====== 典型查询场景 ======

1. 追踪某次任务全链路:
   trace_id:"Task-20260630143000-abc123" | SELECT * ORDER BY __time__

2. 统计各 Agent 执行耗时 P50/P99:
   agent_name:* | SELECT agent_name, approx_percentile(duration_ms, 0.5) AS p50, approx_percentile(duration_ms, 0.99) AS p99 GROUP BY agent_name

3. 告警：最近 5 分钟 ERROR 数量:
   level:ERROR AND __time__ > now() - 300

4. 某任务最近 1 小时成功率:
   task_name:"content_generation" AND __time__ > now() - 3600 | SELECT status, count(*) GROUP BY status

5. 按异常类型聚合排障:
   level:ERROR AND __time__ > now() - 3600 | SELECT error_type, count(*) AS cnt GROUP BY error_type ORDER BY cnt DESC

6. 慢任务检测（耗时 > 阈值）:
   category:task AND duration_ms > 300000 | SELECT task_name, trace_id, duration_ms ORDER BY duration_ms DESC
"""

from __future__ import annotations

from enum import Enum
from typing import Any, Dict, Optional

from pydantic import BaseModel, Field


# ==================== 枚举 ====================


class LogLevel(str, Enum):
    """日志级别"""

    DEBUG = "DEBUG"
    INFO = "INFO"
    WARN = "WARN"
    ERROR = "ERROR"
    FATAL = "FATAL"


class LogCategory(str, Enum):
    """日志分类 —— 按系统边界划分，用于快速过滤"""

    SYSTEM = "system"    # 启动、关闭、心跳
    TASK = "task"        # 任务调度生命周期
    AGENT = "agent"      # Agent 执行（规划、推理、反思）
    TOOL = "tool"        # Tool 调用（HTTP、DB 查询等）
    HTTP = "http"        # 入站 HTTP 请求
    DB = "db"            # 数据库操作
    SECURITY = "security"  # 安全相关（认证、鉴权、告警）


class LogStatus(str, Enum):
    """执行状态"""

    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"
    CANCELLED = "cancelled"
    PENDING = "pending"


# ==================== 结构化日志模型 ====================


class LogRecord(BaseModel):
    """统一日志记录 —— 所有日志推送都必须使用此模型

    使用方式:
        record = LogRecord(
            level=LogLevel.INFO,
            category=LogCategory.TASK,
            event="task_started",
            message="内容生成任务开始执行",
            trace_id="Task-20260630143000-abc123",
            task_name="content_generation",
            extra={"topic": "AI 安全", "word_count": 5000},
        )
        await log_service.log(record)
    """

    # ── 核心字段 ──
    level: LogLevel = Field(default=LogLevel.INFO, description="日志级别")
    category: LogCategory = Field(default=LogCategory.SYSTEM, description="日志分类")
    event: str = Field(default="", description="事件名，如 task_started / agent_think / tool_error")
    message: str = Field(default="", description="人类可读描述，用于全文检索")

    # ── 追踪字段 ──
    trace_id: str = Field(default="", description="分布式追踪 ID，串联完整调用链")
    span_id: str = Field(default="", description="当前 span ID，可选")

    # ── 业务标识 ──
    task_name: str = Field(default="", description="任务名，如 content_generation")
    agent_name: str = Field(default="", description="Agent 名，如 ContentWriter")
    tool_name: str = Field(default="", description="Tool 名，如 WebSearchTool")

    # ── 性能字段 ──
    duration_ms: int = Field(default=0, description="执行耗时（毫秒），用于 P50/P99 分析")

    # ── 状态与异常 ──
    status: LogStatus = Field(default=LogStatus.SUCCESS, description="执行状态")
    error_type: str = Field(default="", description="异常类名，如 TimeoutError")
    error_message: str = Field(default="", description="异常消息全文，用于全文检索")

    # ── 扩展字段 ──
    extra: Dict[str, Any] = Field(default_factory=dict, description="业务自定义 KV，JSON 索引可查询子字段")

    def to_sls_items(self) -> list[tuple[str, str]]:
        """转为阿里云 SLS LogItem 的 contents 格式 [(key, value), ...]

        SLS API 要求所有值都是 string，此方法负责序列化。
        """
        return [
            ("level", self.level.value),
            ("category", self.category.value),
            ("event", self.event),
            ("message", self.message),
            ("trace_id", self.trace_id),
            ("span_id", self.span_id),
            ("task_name", self.task_name),
            ("agent_name", self.agent_name),
            ("tool_name", self.tool_name),
            ("duration_ms", str(self.duration_ms)),
            ("status", self.status.value),
            ("error_type", self.error_type),
            ("error_message", self.error_message),
            # extra 存 JSON 字符串，SLS json 类型索引可解析
            ("extra", json.dumps(self.extra, ensure_ascii=False) if self.extra else "{}"),
        ]


# ==================== SLS 索引配置（文档 / 参考） ====================


SLS_INDEX_CONFIG = {
    "logstore": "long_articles_agentic_aupply",
    "ttl_days": 30,
    "full_text_index": {
        "case_sensitive": False,
        "include_chinese": True,
        "delimiter": ",;=|'\"\t\r\n ",
        "fields": ["message", "error_message"],
    },
    "key_value_index": [
        {"key": "level", "type": "text", "alias": "log_level", "stats": True},
        {"key": "category", "type": "text", "alias": "log_category"},
        {"key": "event", "type": "text"},
        {"key": "trace_id", "type": "text"},
        {"key": "task_name", "type": "text"},
        {"key": "agent_name", "type": "text"},
        {"key": "tool_name", "type": "text"},
        {"key": "duration_ms", "type": "long", "stats": True},
        {"key": "status", "type": "text"},
        {"key": "error_type", "type": "text"},
        {"key": "extra", "type": "json"},
    ],
}


# ── 惰性 json import（避免循环） ──
import json  # noqa: E402