[Log] Add trace log and add loggingInstrumentor tool

[qwes5s5]

FastDeploy 日志与追踪改进方案

一、需求背景

在 RL Profiler 能力建设 过程中，需要支持用户在分析单条数据时能够下钻到具体服务，进一步分析各模块的耗时情况。

目前推理阶段缺乏细粒度的时间打点数据，无法支撑对推理内部阶段的耗时分布查询。因此，需要对 FastDeploy 推理阶段 进行细化划分，并增加日志打点。

此外，现有日志系统存在以下问题：

1. 日志仅包含时间、进程、消息等预定义信息，缺乏上下文链路信息，无法快速定位所属的 Trace 与 Span。

2. 系统可观测性不足，故障排查效率低。

3. 当前日志格式固定，仅支持 msg 参数，缺乏扩展能力。

为了解决无法快速定位的问题，引入 OpenTelemetry LoggingInstrumentor 工具，将 日志（Logs） 与 追踪（Traces） 关联起来，从而提升系统的可观测性与调试能力。

---

二、改动点

1. 新增 Trace Logger

• 新建 trace_logger，并提供获取函数 get_trace_logger。

• 后续所有 FastDeploy 各阶段耗时打点 均通过此 logger 进行。

• 日志输出至独立文件 trace_logger.log。

2. 新增自定义 Formatter

• 创建新的格式化器 CustomColoredFormatter，与 RL Profiler 格式兼容。

• 支持日志扩展字段 extra={"attributes": {...}}。

• 目前在打印时会添加以下关键信息：

  • request_id

  • user_id

  • event

  • stage

• 有助于后续日志聚合与性能分析。

打点示例

trace_logger.info(
    "preprocess end",
    extra={
        "attributes": {
            "request_id": f"{request_id}",
            "user_id": f"{getattr(request, 'user', '')}",
            "event": "POSTPROCESSING_END",
            "stage": "POSTPROCESSING",
        }
    },
)

打印示例（未开启 Trace）

[2025-10-30 06:33:47,246] [INFO    ] (engine_client.py:add_requests:175)
[thread=140634757722560] [thread_name=MainThread] [timestamp=1761806027246]
[request_id=chatcmpl-114514-29ff96da-3825-41a1-acd8-601890942bef]
[user_id=114514] [event=PREPROCESSING_START] [stage=PREPROCESSING]
preprocess is started

---

3. 引入 LoggingInstrumentor

• 接入 OpenTelemetry LoggingInstrumentor。

• 当系统开启 Trace 时，会自动在日志中注入：

  • otel_trace_id

  • otel_span_id

• 修改 Formatter 逻辑，在日志中体现 Trace 相关字段。

打印示例（开启 Trace）

trace_logger：

[2025-10-30 07:47:28,661] [INFO    ] (engine_client.py:add_requests:175)
[otel_trace_id=cea88436e433abb227589130de85a620]
[otel_span_id=f0e8826f89e7f716]
[thread=139858837144000] [thread_name=MainThread] [timestamp=1761810448661]
[request_id=chatcmpl-b6d68f14-4477-4d9d-a6de-bacc60ea782a]
[event=PREPROCESSING_START] [stage=PREPROCESSING]
preprocess is started

api_server_logger：

INFO 2025-10-30 07:47:28,710 21681 serving_chat.py[line:307]
[otel_trace_id=cea88436e433abb227589130de85a620]
[otel_span_id=f0e8826f89e7f716]
Chat Streaming response send_idx 0:
{"id":"chatcmpl-b6d68f14-4477-4d9d-a6de-bacc60ea782a", ...}

---

4. FastDeploy 阶段划分与打点

在 FastDeploy 各主要阶段插入日志打点，以支持耗时分析与追踪。

阶段划分图

打点事件与阶段对应表

事件描述	event 名称	stage 名称
请求前处理开始	PREPROCESSING_START	PREPROCESSING
请求前处理结束	PREPROCESSING_END	PREPROCESSING
请求调度开始	REQUEST_SCHEDULE_START	SCHEDULE
请求排队开始	REQUEST_QUEUE_START	SCHEDULE
请求排队结束	REQUEST_QUEUE_END	SCHEDULE
资源分配开始	RESOURCE_ALLOCATE_START	SCHEDULE
资源分配结束	RESOURCE_ALLOCATE_END	SCHEDULE
请求调度结束	REQUEST_SCHEDULE_END	SCHEDULE
开始推理	INFERENCE_START	PREFILL
首 Token 生成	FIRST_TOKEN_GENERATED	PREFILL
Decode 开始	DECODE_START	DECODE
推理完成	INFERENCE_END	DECODE
后处理开始	POSTPROCESSING_START	POSTPROCESSING
后处理结束	POSTPROCESSING_END	POSTPROCESSING

5. 打点工具类实现

为了规范化和自动化日志追踪信息的记录，定义了以下核心组件：

核心枚举类 (Enums)

这些枚举定义了 FastDeploy 请求处理流程中的标准打点事件和阶段，是实现细粒度追踪的基础。

• LoggingEventName:

  • 作用: 定义了请求处理流程中所有细粒度的打点事件（如 PREPROCESSING_START, INFERENCE_END）。
  • 定位: 最小的耗时分析单元。

• StageName:

  • 作用: 定义了请求处理流程中所有宏观的主阶段（如 PREPROCESSING, SCHEDULE, DECODE）。
  • 定位: 高层级的流程划分和性能概览。

映射关系

• EVENT_TO_STAGE_MAP:
  • 作用: 将每一个细粒度的 LoggingEventName 映射到其所属的 StageName。
  • 价值: 确保日志在记录 Event 时，能够自动附加其所属的 Stage 信息，从而简化打点流程并便于后续的日志查询和聚合分析。

追踪日志函数 (trace_logging)

这是实际执行日志打点的工具函数。它封装了复杂的日志 extra 字段构造逻辑，确保日志记录的简洁性和上下文完整性。

• 作用: 记录任务的跟踪日志信息，自动注入 request_id, user_id, event 和 stage 等关键字段。
• 精简流程: 只需传入 msg、任务对象（dict 或 object 形式）和 event 即可完成完整的打点。

---

from enum import Enum
from fastdeploy.utils import trace_logger

class LoggingEventName(Enum):
    """
    表示系统中的各种事件点，用于细粒度耗时追踪。
    """
    PREPROCESSING_START = "PREPROCESSING_START"
    PREPROCESSING_END = "PREPROCESSING_END"
    REQUEST_SCHEDULE_START = "REQUEST_SCHEDULE_START"
    REQUEST_QUEUE_START = "REQUEST_QUEUE_START"
    REQUEST_QUEUE_END = "REQUEST_QUEUE_END"
    RESOURCE_ALLOCATE_START = "RESOURCE_ALLOCATE_START"
    RESOURCE_ALLOCATE_END = "RESOURCE_ALLOCATE_END"
    REQUEST_SCHEDULE_END = "REQUEST_SCHEDULE_END"
    INFERENCE_START = "INFERENCE_START"
    FIRST_TOKEN_GENERATED = "FIRST_TOKEN_GENERATED"
    DECODE_START = "DECODE_START"
    INFERENCE_END = "INFERENCE_END"
    POSTPROCESSING_START = "POSTPROCESSING_START"
    POSTPROCESSING_END = "POSTPROCESSING_END"


class StageName(Enum):
    """
    表示请求处理流程中的主要阶段（宏观阶段划分）。
    """
    PREPROCESSING = "PREPROCESSING"
    SCHEDULE = "SCHEDULE"
    PREFILL = "PREFILL"
    DECODE = "DECODE"
    POSTPROCESSING = "POSTPROCESSING"


EVENT_TO_STAGE_MAP = {
    # 自动将 Event 映射到其所属的 Stage
    LoggingEventName.PREPROCESSING_START: StageName.PREPROCESSING,
    LoggingEventName.PREPROCESSING_END: StageName.PREPROCESSING,
    # ... (其他映射)
    LoggingEventName.INFERENCE_START: StageName.PREFILL,
    LoggingEventName.INFERENCE_END: StageName.DECODE,
    # ...
}


def trace_logging(msg: str, task: [dict, object], event: LoggingEventName):
    """
    记录任务的跟踪日志信息。自动提取 request_id 和 user_id，并根据 event 自动推断 stage。
    
    Args:
        msg (str): 日志消息文本。
        task (dict/object): 待记录的任务对象（需包含 request_id 和 user 属性/键）。
        event (LoggingEventName): 当前发生的事件点。
    """
    try:
        if isinstance(task, dict):
            req_id = task.get('request_id', 'N/A')
            user_id = task.get('user', '')
        elif task is not None:
            req_id = getattr(task, 'request_id', 'N/A')
            user_id = getattr(task, 'user', '')
        else:
            return # task为空则不记录

        trace_logger.info(
            msg,
            extra={
                "attributes": {
                    "request_id": f"{req_id}",
                    "user_id": f"{user_id}",
                    "event": event.value,
                    "stage": EVENT_TO_STAGE_MAP.get(event).value,
                }
            },
        )
    except Exception as e:
        # 捕获异常但不中断主流程，确保日志记录失败不影响推理服务
        # print(f"Trace logging failed: {e}") 
        pass

---

[paddle-bot]

Thanks for your contribution!

[qwes5s5]

 /re-run run_tests_with_coverage

[qwes5s5]

/re-run run_tests_with_coverage