xiaoxia-saas/apps/worker/video_processing/render_metrics.py

"""渲染引擎灰度观测指标。

提供 Prometheus 指标埋点，用于灰度发布期间观测新旧引擎的：
  - 任务成功率
  - 耗时分布
  - 错误类型分布
  - stream_copy 命中率

使用方式（任务入口）：
    with render_task_metrics(engine="unified", task_type="compose_video"):
        # 执行渲染任务
        result = do_render()

stream_copy 埋点：
    record_stream_copy(result="hit", reason="成功")
    record_stream_copy(result="miss", reason="编码不匹配")
    record_stream_copy(result="fallback", reason="失败回退")

Worker 进程启动时需调用 start_metrics_server(port) 暴露 /metrics 端点。
"""

from __future__ import annotations

import logging
import threading
import time
from contextlib import contextmanager
from typing import Iterator, Optional

from prometheus_client import REGISTRY, Counter, Gauge, Histogram, start_http_server

logger = logging.getLogger(__name__)


# ── 指标定义 ──────────────────────────────────────────────────────────────────

# 渲染任务总数（按引擎、任务类型、状态区分）
RENDER_TASK_TOTAL = Counter(
    "render_task_total",
    "Total number of render tasks",
    ["engine", "task_type", "status"],
    registry=REGISTRY,
)

# 渲染任务耗时直方图（按引擎、任务类型区分）
# buckets 覆盖从秒级到小时级，适配短视频到长视频的渲染场景
RENDER_TASK_DURATION_SECONDS = Histogram(
    "render_task_duration_seconds",
    "Render task duration in seconds",
    ["engine", "task_type"],
    buckets=(1, 5, 10, 30, 60, 120, 300, 600, 900, 1800, 3600, 7200),
    registry=REGISTRY,
)

# 渲染错误统计（按引擎、任务类型、错误类型区分）
RENDER_ERROR_TOTAL = Counter(
    "render_error_total",
    "Total number of render errors",
    ["engine", "task_type", "error_type"],
    registry=REGISTRY,
)

# stream_copy 命中统计（仅新引擎有意义）
RENDER_STREAM_COPY_TOTAL = Counter(
    "render_stream_copy_total",
    "Total number of stream copy attempts and results",
    ["result", "reason"],
    registry=REGISTRY,
)

# 当前正在执行的渲染任务数
RENDER_ACTIVE_TASKS = Gauge(
    "render_active_tasks",
    "Number of render tasks currently in progress",
    ["engine", "task_type"],
    registry=REGISTRY,
)


# ── Metrics Server ───────────────────────────────────────────────────────────

_metrics_server_started = False
_metrics_server_lock = threading.Lock()


def start_metrics_server(port: int = 9101) -> None:
    """启动 Prometheus metrics HTTP 服务器。

    在 worker 进程启动时调用一次即可，多进程环境下每个 worker 进程
    会启动自己的 metrics server（需配置不同端口或使用进程号偏移）。

    Args:
        port: metrics 服务端口，默认 9101
    """
    global _metrics_server_started

    with _metrics_server_lock:
        if _metrics_server_started:
            return

        try:
            start_http_server(port)
            _metrics_server_started = True
            logger.info("Render metrics server started on port %d", port)
        except OSError as e:
            # 端口已占用可能是多 worker 进程场景，记录警告不阻断
            logger.warning("Failed to start metrics server on port %d: %s", port, e)


# ── 任务级埋点 ───────────────────────────────────────────────────────────────

@contextmanager
def render_task_metrics(engine: str, task_type: str) -> Iterator[None]:
    """渲染任务指标上下文管理器。

    自动记录：任务开始（活跃数+1）、任务结束（耗时 + 状态 + 活跃数-1）。

    Args:
        engine: 渲染引擎类型，"legacy" 或 "unified"
        task_type: 任务类型，"compose_video" / "edit_plan" / "generate_video"

    Usage:
        with render_task_metrics(engine="unified", task_type="compose_video"):
            result = do_render()
            # 正常退出 = success
            # 抛异常 = failure（会记录 error_type）
    """
    RENDER_ACTIVE_TASKS.labels(engine=engine, task_type=task_type).inc()
    start_time = time.perf_counter()
    status = "success"

    try:
        yield
    except Exception as e:
        status = "failure"
        error_type = _classify_error(e)
        RENDER_ERROR_TOTAL.labels(
            engine=engine,
            task_type=task_type,
            error_type=error_type,
        ).inc()
        raise
    finally:
        duration = time.perf_counter() - start_time
        RENDER_TASK_TOTAL.labels(
            engine=engine,
            task_type=task_type,
            status=status,
        ).inc()
        RENDER_TASK_DURATION_SECONDS.labels(
            engine=engine,
            task_type=task_type,
        ).observe(duration)
        RENDER_ACTIVE_TASKS.labels(engine=engine, task_type=task_type).dec()


def record_stream_copy(result: str, reason: str) -> None:
    """记录 stream_copy 命中/跳过/回退情况。

    Args:
        result: "hit"（命中直通） / "miss"（条件不满足跳过） / "fallback"（失败回退）
        reason: 具体原因，如"编码不匹配"、"分辨率不同"、"成功"、"ffmpeg失败"等
    """
    RENDER_STREAM_COPY_TOTAL.labels(result=result, reason=reason).inc()


# ── 辅助函数 ─────────────────────────────────────────────────────────────────

def _classify_error(exc: BaseException) -> str:
    """将异常分类为标准错误类型。

    用于 error_type 标签，控制指标基数不要爆炸。
    """
    import subprocess

    if isinstance(exc, subprocess.TimeoutExpired):
        return "timeout"
    if isinstance(exc, subprocess.CalledProcessError):
        return "ffmpeg_error"
    if isinstance(exc, ValueError):
        return "validation"
    if isinstance(exc, (OSError, IOError)):
        return "io_error"

    exc_name = type(exc).__name__
    # 常见的 OSS / 网络相关异常
    if "oss" in exc_name.lower() or "storage" in exc_name.lower():
        return "oss_error"
    if "timeout" in exc_name.lower():
        return "timeout"

    return "unknown"


def classify_stream_copy_miss_reason(reason: str) -> str:
    """将 stream_copy 未命中原因归一化为标准分类。

    控制 reason 标签基数，避免爆炸。
    """
    if "编码" in reason or "codec" in reason.lower():
        return "编码不匹配"
    if "分辨率" in reason or "width" in reason.lower() or "height" in reason.lower():
        return "分辨率不匹配"
    if "帧率" in reason or "fps" in reason.lower():
        return "帧率不匹配"
    if "字幕" in reason or "ass" in reason.lower():
        return "有字幕叠加"
    if "像素格式" in reason or "pix_fmt" in reason.lower():
        return "像素格式不匹配"
    if "trim" in reason or "时长" in reason:
        return "裁剪不支持"
    if "多clip" in reason or "多片段" in reason or "转场" in reason:
        return "多片段不支持"
    return "其他"