146 lines
4.4 KiB
Python
146 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""latency —— 时延打点与 SLA 判定（M11b）。
|
||
|
||
承诺（m11b_config）：裁决 ≤200ms、广播 ≤300ms、端到端 ≤1000ms、轮询兜底 3000ms。
|
||
本模块只做纯内存统计（可离线断言），不依赖 DB；真实数值由 selftest 连库实测填入报告。
|
||
"""
|
||
|
||
import time
|
||
from collections import deque
|
||
|
||
from .m11b_config import (POLL_FALLBACK_MS, SLA_ADJUDICATE_MS, SLA_BROADCAST_MS,
|
||
SLA_END_TO_END_MS)
|
||
|
||
STAGES = ("adjudicate", "broadcast", "end_to_end")
|
||
|
||
|
||
def percentile(values, p):
|
||
if not values:
|
||
return None
|
||
s = sorted(values)
|
||
idx = min(len(s) - 1, int(round((p / 100.0) * (len(s) - 1))))
|
||
return round(s[idx], 3)
|
||
|
||
|
||
class Timer(object):
|
||
"""with Timer() as t: ... ; t.ms —— 毫秒级计时。"""
|
||
|
||
def __init__(self):
|
||
self._t0 = None
|
||
self.ms = None
|
||
|
||
def __enter__(self):
|
||
self._t0 = time.perf_counter()
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
self.ms = round((time.perf_counter() - self._t0) * 1000.0, 3)
|
||
return False
|
||
|
||
|
||
class LatencyTracker(object):
|
||
"""各阶段时延样本环形记录 + SLA 越界计数。"""
|
||
|
||
def __init__(self, maxlen=500):
|
||
self._samples = dict((s, deque(maxlen=maxlen)) for s in STAGES)
|
||
self._breach = dict((s, 0) for s in STAGES)
|
||
|
||
@staticmethod
|
||
def sla_for(stage):
|
||
return {"adjudicate": SLA_ADJUDICATE_MS,
|
||
"broadcast": SLA_BROADCAST_MS,
|
||
"end_to_end": SLA_END_TO_END_MS}[stage]
|
||
|
||
def record(self, stage, ms):
|
||
if stage not in self._samples:
|
||
raise KeyError("unknown stage %r" % stage)
|
||
ms = round(float(ms), 3)
|
||
self._samples[stage].append(ms)
|
||
breach = ms > self.sla_for(stage)
|
||
if breach:
|
||
self._breach[stage] += 1
|
||
return breach
|
||
|
||
def summary(self):
|
||
out = {}
|
||
for s in STAGES:
|
||
vals = list(self._samples[s])
|
||
n = len(vals)
|
||
out[s] = {
|
||
"sla_ms": self.sla_for(s),
|
||
"samples": n,
|
||
"avg_ms": round(sum(vals) / n, 3) if n else None,
|
||
"p95_ms": percentile(vals, 95),
|
||
"p99_ms": percentile(vals, 99),
|
||
"max_ms": round(max(vals), 3) if n else None,
|
||
"breaches": self._breach[s],
|
||
}
|
||
out["poll_fallback_ms"] = POLL_FALLBACK_MS
|
||
return out
|
||
|
||
def all_within_sla(self):
|
||
return all(self._breach[s] == 0 for s in STAGES)
|
||
|
||
|
||
_TRACKER = None
|
||
|
||
|
||
def get_tracker():
|
||
global _TRACKER
|
||
if _TRACKER is None:
|
||
_TRACKER = LatencyTracker()
|
||
return _TRACKER
|
||
|
||
|
||
def record(stage, ms):
|
||
return get_tracker().record(stage, ms)
|
||
|
||
|
||
def summary():
|
||
return get_tracker().summary()
|
||
|
||
|
||
def self_check():
|
||
msgs, ok = [], True
|
||
t = LatencyTracker()
|
||
if t.record("adjudicate", 120.0) is not False:
|
||
ok = False
|
||
msgs.append("latency FAIL：120ms 不应判越界（SLA 200ms）")
|
||
else:
|
||
msgs.append("latency PASS：裁决 120ms ≤ 200ms 未越界")
|
||
if t.record("adjudicate", 450.0) is not True:
|
||
ok = False
|
||
msgs.append("latency FAIL：450ms 应判越界")
|
||
else:
|
||
msgs.append("latency PASS：裁决 450ms > 200ms 记越界")
|
||
t.record("broadcast", 40.0)
|
||
t.record("end_to_end", 620.0)
|
||
s = t.summary()
|
||
if s["adjudicate"]["breaches"] != 1 or s["adjudicate"]["samples"] != 2:
|
||
ok = False
|
||
msgs.append("latency FAIL：summary %r" % s["adjudicate"])
|
||
else:
|
||
msgs.append("latency PASS：summary avg=%s p95=%s breaches=%d"
|
||
% (s["adjudicate"]["avg_ms"], s["adjudicate"]["p95_ms"],
|
||
s["adjudicate"]["breaches"]))
|
||
if s["poll_fallback_ms"] != 3000:
|
||
ok = False
|
||
msgs.append("latency FAIL：轮询兜底 %s 应为 3000" % s["poll_fallback_ms"])
|
||
else:
|
||
msgs.append("latency PASS：轮询兜底 3000ms 已配置")
|
||
if percentile([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 95) is None:
|
||
ok = False
|
||
msgs.append("latency FAIL：percentile 空值")
|
||
else:
|
||
msgs.append("latency PASS：p95=%s" % percentile(range(1, 11), 95))
|
||
with Timer() as tm:
|
||
time.sleep(0.005)
|
||
if not (1.0 <= tm.ms < 2000):
|
||
ok = False
|
||
msgs.append("latency FAIL：Timer 计时 %s" % tm.ms)
|
||
else:
|
||
msgs.append("latency PASS：Timer 实测 %.1fms" % tm.ms)
|
||
if ok:
|
||
msgs.append("latency.self_check PASS")
|
||
return ok, msgs
|