灰度、Feature Flag 与回滚

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

mkdir -p l8-release
cd l8-release

"""演示稳定分组、功能开关与固定观测窗口的发布决策。"""
from hashlib import sha256
from decimal import Decimal

MAX_BAD_RATE = Decimal("0.02")
# 观测窗口固定100次新路径请求，有3次业务失败。
WINDOW = [True] * 97 + [False] * 3

def enabled(user_id, percent, kill=False):
    if not 0 <= percent <= 100:
        raise ValueError("灰度比例必须在0到100之间")
    if kill:
        return False
    # 相同用户始终落入同一桶；加大比例保留已进入灰度的用户。
    bucket = int.from_bytes(sha256(user_id.encode()).digest()[:4], "big") % 100
    return bucket < percent

def decision(window, threshold, minimum=100):
    if not Decimal("0") <= threshold <= Decimal("1"):
        raise ValueError("失败率阈值必须在0到1之间")
    if len(window) < minimum:
        return "保持灰度", None
    rate = Decimal(sum(not good for good in window)) / len(window)
    return ("关闭开关并回滚路由" if rate > threshold else "扩大到25%"), rate

def main():
    users = [f"user-{i}" for i in range(100)]
    cohort = sum(enabled(user, 5) for user in users)
    print(f"09:00 开关5% / 固定100名用户进入新路径{cohort}名")
    print("09:05 新路径请求100次 / 失败3次")
    action, rate = decision(WINDOW, MAX_BAD_RATE)
    print(f"失败率: {rate * 100:.2f}% / 阈值: {MAX_BAD_RATE * 100:.2f}%")
    print(f"09:06 决策: {action}")
    print("紧急关闭后新路径人数:", sum(enabled(user, 5, kill=True) for user in users))

if __name__ == "__main__":
    main()

"""验证确定性分组、扩容单调性、关停优先与观测门槛。"""
from decimal import Decimal as D
import unittest
from canary_release import enabled, decision, WINDOW

class ReleaseTests(unittest.TestCase):
    def test_same_user_and_monotonic(self):
        for i in range(100):
            user = f"user-{i}"
            self.assertEqual(enabled(user, 5), enabled(user, 5))
            self.assertFalse(enabled(user, 5) and not enabled(user, 25))

    def test_kill_and_percentage_edges(self):
        self.assertFalse(enabled("alice", 100, kill=True))
        self.assertFalse(enabled("alice", 0))
        self.assertTrue(enabled("alice", 100))
        with self.assertRaises(ValueError):
            enabled("alice", 101)

    def test_insufficient_window(self):
        self.assertEqual(decision([False] * 99, D("0.02")), ("保持灰度", None))

    def test_failure_and_changed_threshold(self):
        self.assertEqual(decision(WINDOW, D("0.02"))[0], "关闭开关并回滚路由")
        self.assertEqual(decision(WINDOW, D("0.04"))[0], "扩大到25%")

    def test_exact_threshold(self):
        self.assertEqual(decision([True] * 98 + [False] * 2, D("0.02"))[0], "扩大到25%")

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

ls
python -X utf8 canary_release.py
python -X utf8 -m unittest -v test_canary_release.py

09:00 开关5% / 固定100名用户进入新路径7名
09:05 新路径请求100次 / 失败3次
失败率: 3.00% / 阈值: 2.00%
09:06 决策: 关闭开关并回滚路由
紧急关闭后新路径人数: 0

python -X utf8 canary_release.py