"""压力指标采集与计算（任务书 §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, }