SLO、SLA 与错误预算

New-Item -ItemType Directory -Force l8-slo
cd l8-slo

mkdir -p l8-slo
cd l8-slo

"""从预设请求记录计算用户可感知成功率和错误预算。"""
from decimal import Decimal
from math import ceil

TARGET = Decimal("0.99")
# 每项为（服务是否成功完成、端到端毫秒数）；共1000条固定记录。
REQUESTS = [(True, 400)] * 950 + [(True, 800)] * 30 + [(False, 3000)] * 20

def summarize(requests, target):
    if not Decimal("0") < target <= Decimal("1"):
        raise ValueError("目标必须大于0且不超过1")
    if not requests:
        raise ValueError("窗口内没有有效请求")
    total = len(requests)
    # 业务成功且在2秒内完成才算好请求，超时也消耗错误预算。
    good = sum(ok and elapsed <= 2000 for ok, elapsed in requests)
    bad = total - good
    allowed = Decimal(total) * (1 - target)
    remaining = allowed - bad
    # 用最近秩定义p95：排序后第 ceil(0.95*N) 项。
    latencies = sorted(elapsed for _, elapsed in requests)
    p95 = latencies[ceil(0.95 * total) - 1]
    return total, good, bad, allowed, remaining, p95

def main():
    total, good, bad, allowed, remaining, p95 = summarize(REQUESTS, TARGET)
    print(f"窗口: {total} / 好请求: {good} / 坏请求: {bad}")
    print(f"SLI: {Decimal(good) / total * 100:.2f}% / SLO: {TARGET * 100:.2f}%")
    print(f"允许坏请求: {allowed:.0f} / 剩余预算: {remaining:.0f}")
    print(f"p95: {p95}毫秒")
    print("发布策略:", "冻结非修复发布" if remaining < 0 else "允许受控发布")

if __name__ == "__main__":
    main()

"""检验窗口、成功判据、最近秩和预算耗尽的边界。"""
from decimal import Decimal as D
import unittest
from error_budget import REQUESTS, summarize

class BudgetTests(unittest.TestCase):
    def test_baseline(self):
        self.assertEqual(summarize(REQUESTS, D("0.99")),
                         (1000, 980, 20, D(10), D(-10), 400))

    def test_revised_target(self):
        self.assertEqual(summarize(REQUESTS, D("0.97"))[4], D(10))

    def test_budget_exactly_zero(self):
        self.assertEqual(summarize([(True, 1)] * 99 + [(False, 1)], D("0.99"))[4], D(0))

    def test_latency_and_business_success(self):
        self.assertEqual(summarize([(True, 2000), (True, 2001), (False, 10)], D("0.9"))[1:3],
                         (1, 2))

    def test_empty_and_invalid(self):
        with self.assertRaises(ValueError):
            summarize([], D("0.99"))
        for target in (D(0), D("1.1")):
            with self.assertRaises(ValueError):
                summarize(REQUESTS, target)

if __name__ == "__main__":
    unittest.main()

ls
python -X utf8 error_budget.py
python -X utf8 -m unittest -v test_error_budget.py

窗口: 1000 / 好请求: 980 / 坏请求: 20
SLI: 98.00% / SLO: 99.00%
允许坏请求: 10 / 剩余预算: -10
p95: 400毫秒
发布策略: 冻结非修复发布

python -X utf8 error_budget.py