AiDOPWarehouse/tools/integration-simulator/load/metrics.py

"""压力指标采集与计算（任务书 §17）。

只采集第三方侧可观察指标：请求量、批次、行数、成功率、QPS、行吞吐、
延迟分布（P50/P95/P99）、HTTP 错误分类、超时数、限流数、首次失败时间、起止时间。
Ai-DOP CPU/内存/队列/数据库资源由测试人员在现有监控工具观察（§17 末句）。
"""
from __future__ import annotations

import threading
import time

# §17 指标快照字段（导出报告按此列全量输出）
METRIC_FIELDS = (
    ("totalRequests", "总请求数"),
    ("totalBatches", "总批次数"),
    ("totalRows", "总数据行数"),
    ("success", "成功数"),
    ("failed", "失败数"),
    ("successRate", "成功率"),
    ("actualQps", "实际 QPS"),
    ("rowThroughput", "行吞吐（行/秒）"),
    ("avgMs", "平均耗时（ms）"),
    ("p50Ms", "P50（ms）"),
    ("p95Ms", "P95（ms）"),
    ("p99Ms", "P99（ms）"),
    ("maxMs", "最大耗时（ms）"),
    ("httpErrors", "HTTP 错误分类"),
    ("timeouts", "超时数"),
    ("rateLimited", "限流数"),
    ("firstFailureAt", "首次失败时间"),
    ("startedAt", "开始时间"),
    ("finishedAt", "结束时间"),
)


def percentile(sorted_values: list[float], p: float) -> float:
    """线性插值百分位（P50/P95/P99）。空列表返回 0.0。"""
    if not sorted_values:
        return 0.0
    if p <= 0:
        return sorted_values[0]
    if p >= 100:
        return sorted_values[-1]
    k = (len(sorted_values) - 1) * p / 100.0
    lo = int(k)
    hi = min(lo + 1, len(sorted_values) - 1)
    return sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * (k - lo)


class LoadMetrics:
    """线程安全的压力指标累计器。

    QPS 口径：最近 window_seconds 内完成的请求数 / 窗口秒数（滚动窗口，
    长任务结束后与实际总吞吐可解释比对）。
    """

    def __init__(self, window_seconds: float = 60.0):
        self.window_seconds = max(1.0, window_seconds)
        self.started_at: str | None = None
        self.finished_at: str | None = None
        self.total_requests = 0
        self.total_batches = 0
        self.total_rows = 0
        self.success = 0
        self.failed = 0
        self.timeouts = 0
        self.rate_limited = 0
        self.http_errors: dict[int, int] = {}
        self.latencies: list[float] = []
        self.first_failure_at: str | None = None
        self._completion_times: list[float] = []
        self._final_qps: float | None = None
        self._final_throughput: float | None = None
        self._lock = threading.Lock()

    def reset(self) -> None:
        """清空全部累计（新一轮压测开始前由 mark_start 调用）。"""
        self.started_at = None
        self.finished_at = None
        self.total_requests = 0
        self.total_batches = 0
        self.total_rows = 0
        self.success = 0
        self.failed = 0
        self.timeouts = 0
        self.rate_limited = 0
        self.http_errors = {}
        self.latencies = []
        self.first_failure_at = None
        self._completion_times = []
        self._final_qps = None
        self._final_throughput = None

    def mark_start(self) -> None:
        with self._lock:
            # 单例跨轮复用：新一轮压测开始即重置上一轮累计
            self.reset()
            self.started_at = time.strftime("%Y-%m-%dT%H:%M:%S")

    def mark_finish(self) -> None:
        with self._lock:
            if self.finished_at is None:
                self.finished_at = time.strftime("%Y-%m-%dT%H:%M:%S")
                # 结束后冻结滚动指标：QPS/行吞吐定格在结束时刻，不再随时间衰减
                now = time.monotonic()
                self._final_qps = self._actual_qps_locked(now)
                self._final_throughput = self._row_throughput_locked()

    def record(self, status: int, latency_s: float, rows: int = 0, batches: int = 1) -> None:
        """记录一次请求/批次结果。status：0=超时/不可达，429=限流，其余 HTTP。"""
        now = time.monotonic()
        with self._lock:
            self.total_requests += 1
            self.total_batches += batches
            self.total_rows += max(0, rows)
            self.latencies.append(max(0.0, latency_s))
            self._completion_times.append(now)
            if status == 0:
                self.timeouts += 1
                self.failed += 1
            elif status == 429:
                self.rate_limited += 1
                self.failed += 1
            elif 200 <= status < 300 or status == 202:
                self.success += 1
            else:
                self.failed += 1
                self.http_errors[status] = self.http_errors.get(status, 0) + 1
            if (status == 0 or status >= 400) and self.first_failure_at is None:
                self.first_failure_at = time.strftime("%Y-%m-%dT%H:%M:%S")

    # -- 计算 ----------------------------------------------------------------

    def _pruned_completion_times(self, now: float) -> list[float]:
        cutoff = now - self.window_seconds
        return [t for t in self._completion_times if t >= cutoff]

    def _actual_qps_locked(self, now: float) -> float:
        recent = self._pruned_completion_times(now)
        if not recent:
            return 0.0
        span = now - recent[0]
        if span <= 0:
            return 0.0
        return round(len(recent) / span, 2)

    def actual_qps(self) -> float:
        now = time.monotonic()
        with self._lock:
            return self._actual_qps_locked(now)

    def _row_throughput_locked(self) -> float:
        if not self.started_at or self.total_rows == 0:
            return 0.0
        end = time.strptime(self.finished_at, "%Y-%m-%dT%H:%M:%S") if self.finished_at else time.localtime()
        start = time.strptime(self.started_at, "%Y-%m-%dT%H:%M:%S")
        elapsed = max(1e-9, time.mktime(end) - time.mktime(start))
        return round(self.total_rows / elapsed, 2)

    def row_throughput(self) -> float:
        with self._lock:
            return self._row_throughput_locked()

    # -- 快照 ----------------------------------------------------------------

    def snapshot(self) -> dict:
        with self._lock:
            latencies = sorted(self.latencies)
            total = self.total_requests
            success_rate = round(self.success / total, 4) if total else 0.0
            # 持锁内用 *_locked 计算，避免重入锁死（Lock 非重入）
            now = time.monotonic()
            frozen = self.finished_at is not None
            return {
                "totalRequests": total,
                "totalBatches": self.total_batches,
                "totalRows": self.total_rows,
                "success": self.success,
                "failed": self.failed,
                "successRate": success_rate,
                "actualQps": (self._final_qps if frozen and self._final_qps is not None
                              else self._actual_qps_locked(now)),
                "rowThroughput": (self._final_throughput
                                  if frozen and self._final_throughput is not None
                                  else self._row_throughput_locked()),
                "avgMs": round(sum(latencies) / len(latencies) * 1000, 1) if latencies else 0.0,
                "p50Ms": round(percentile(latencies, 50) * 1000, 1),
                "p95Ms": round(percentile(latencies, 95) * 1000, 1),
                "p99Ms": round(percentile(latencies, 99) * 1000, 1),
                "maxMs": round(max(latencies) * 1000, 1) if latencies else 0.0,
                "httpErrors": dict(sorted(self.http_errors.items())),
                "timeouts": self.timeouts,
                "rateLimited": self.rate_limited,
                "firstFailureAt": self.first_failure_at,
                "startedAt": self.started_at,
                "finishedAt": self.finished_at,
            }