API 灰度判定与回滚引擎（专业版）

输入 / 输出

输入：放量阶梯、错误率与 P99 余量、最小样本；灰度与基线各自的指标
（calls/errors/p99）；用于分流的请求哈希。
输出：判定结果 PROMOTE / CONTINUE / ROLLBACK 及原因、当前灰度权重、请求是否走灰度。

使用示例

from canary_release import CanaryRelease

cr = CanaryRelease(stages=(5, 20, 50, 100), error_margin=0.02, min_sample=100)
if cr.route_to_canary(roll=hash_of_request):
    ...  # 走新版本
decision, reasons = cr.evaluate(canary_metrics, baseline_metrics)
# ROLLBACK -> 自动切回基线；CONTINUE -> cr.advance_stage()；PROMOTE -> 全量

引擎代码：canary_release.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""API 灰度发布（金丝雀）计算引擎（专业版核心）
新版本按权重小比例放量 -> 指标对比基线 -> 健康则逐级提量直至全量(PROMOTE)，
劣化超阈值则自动回滚(ROLLBACK)，否则继续观察(CONTINUE)。
"""

CONTINUE, PROMOTE, ROLLBACK = "CONTINUE", "PROMOTE", "ROLLBACK"


class CanaryRelease:
    def __init__(self, stages=(5, 20, 50, 100), error_margin=0.02,
                 latency_p99_margin=0.10, min_sample=100, target_weight=100):
        # stages: 灰度放量阶梯(%)；error_margin: 错误率允许绝对上浮
        # latency_p99_margin: P99延迟允许相对上浮；min_sample: 判定所需最小灰度样本
        self.stages = list(stages)
        self.error_margin = error_margin
        self.latency_p99_margin = latency_p99_margin
        self.min_sample = min_sample
        self.target_weight = target_weight
        self.stage_idx = 0

    def current_weight(self):
        return self.stages[min(self.stage_idx, len(self.stages) - 1)]

    def route_to_canary(self, roll):
        """roll in [0,100)，判定本次请求是否走灰度"""
        return roll < self.current_weight()

    def advance_stage(self):
        if self.stage_idx < len(self.stages) - 1:
            self.stage_idx += 1
        return self.current_weight()

    def evaluate(self, canary, baseline):
        """canary/baseline: {'calls':int,'errors':int,'p99':float}
        返回 (decision, reasons[])"""
        reasons = []
        calls = canary.get("calls", 0)
        if calls < self.min_sample:
            return CONTINUE, ["灰度样本 %d < min_sample %d，样本不足继续观察" % (calls, self.min_sample)]
        c_err = canary["errors"] / calls
        b_calls = max(baseline.get("calls", 1), 1)
        b_err = baseline.get("errors", 0) / b_calls
        if c_err > b_err + self.error_margin:
            reasons.append("灰度错误率 %.3f > 基线 %.3f + 余量 %.3f" % (c_err, b_err, self.error_margin))
        if baseline.get("p99"):
            rel = (canary["p99"] - baseline["p99"]) / baseline["p99"]
            if rel > self.latency_p99_margin:
                reasons.append("灰度P99相对上浮 %.1f%% > 余量 %.1f%%" % (rel * 100, self.latency_p99_margin * 100))
        if reasons:
            return ROLLBACK, reasons
        if self.current_weight() >= self.target_weight:
            return PROMOTE, ["指标健康且已达目标权重 %d%%，全量发布" % self.target_weight]
        return CONTINUE, ["指标健康，可提量至下一阶梯"]

测试（3 正例 + 1 边界，实跑全过）

硬性规则

1. 分流应保证同一用户稳定命中同一版本（用稳定哈希，不用每次随机）。
2. 样本不足 min_sample 不回滚也不全量；错误率/P99 越界必须自动回滚并验证基线。
3. 放量前确认新旧版本的数据（库表/缓存）兼容。
4. 阈值按业务容错度设定；判定与提量全过程留指标快照与日志。