Flagger Webhook에 Alpha Spending Sequential Testing 구현하기 — 통계적 조기 탈출로 Canary 롤백을 최대 66% 단축하는 방법

import numpy as np
 
def peeking_inflation(alpha: float, num_peeks: int) -> float:
    """반복 검정 시 실제 Type I 오류율 추정 (Bonferroni 상한 — 실제보다 보수적)"""
    # 각 검정이 독립적이라 가정한 보수적 상한값
    # 실제 상관된 관측값에서는 이보다 낮게 나옴 (12회 peek 시 약 0.30~0.35 수준)
    return 1 - (1 - alpha) ** num_peeks
 
# 5분 간격으로 60분 동안 12번 들여다보면
print(peeking_inflation(0.05, 12))  # ≈ 0.46 (Bonferroni 상한)
# 실제 시뮬레이션 기반 추정치는 약 0.30~0.35 수준이지만,
# 어느 쪽이든 목표 오류율 0.05를 대폭 초과한다는 사실은 변하지 않는다

import numpy as np
from scipy import stats
 
def obrien_fleming(tau: float, alpha: float = 0.05) -> float:
    """O'Brien-Fleming 누적 Alpha Spending"""
    if tau <= 0:
        return 0.0
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    return 2 * (1 - stats.norm.cdf(z_alpha / np.sqrt(tau)))
 
def pocock(tau: float, alpha: float = 0.05) -> float:
    """Pocock 누적 Alpha Spending"""
    return alpha * np.log(1 + (np.e - 1) * tau)
 
# τ=0.1 (10% 진행) 시점에서 각 방법이 소비하는 누적 alpha
print(f"O'Brien-Fleming @ τ=0.1: {obrien_fleming(0.1):.6f}")  # ≈ 0.000016
print(f"Pocock          @ τ=0.1: {pocock(0.1):.6f}")           # ≈ 0.017

[Flagger Canary Controller]
        │
        │ 매 interval마다 webhook POST
        ▼
[Sequential Test Service]  ←─── [Prometheus / Metrics Store]
        │                              (현재 canary 메트릭)
        │
        │ 1. 현재 τ 계산 (수집 샘플 / 목표 샘플)
        │ 2. incremental_alpha 계산 (이번 분석의 α 예산)
        │ 3. 임계 Z-값 산출
        │ 4. 실제 Z-통계량과 비교
        ▼
   |Z| > boundary?
   ├── YES → HTTP 400 → Flagger 실패 카운트 증가 → threshold 초과 시 자동 롤백
   └── NO  → HTTP 200 → Canary 계속 진행

# canary.yaml
apiVersion: flagger.app/v1beta1
kind: Canary
metadata:
  name: my-service
  namespace: production
spec:
  targetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: my-service
  progressDeadlineSeconds: 600
  service:
    port: 80
  analysis:
    interval: 1m          # 매 1분 = 한 번의 중간 분석
    threshold: 3          # 3회 연속 실패 시 롤백
    maxWeight: 50         # 최대 50% 트래픽
    stepWeight: 10        # 매 iteration 10% 증가
    metrics:
      - name: request-success-rate
        thresholdRange:
          min: 99
        interval: 1m
    webhooks:
      - name: sequential-test-gate
        type: pre-rollout   # 트래픽 증가 전 게이팅
        url: http://seq-test-service.monitoring/check
        timeout: 30s
        metadata:
          alpha: "0.05"
          method: "obrien_fleming"
          # Flagger metadata 필드는 문자열만 허용하므로 숫자도 문자열로 전달
          target_samples: "10000"
          metric: "error_rate"
          canary_name: "my-service"

# seq_test_service.py — 핵심 통계 로직
import numpy as np
from scipy import stats
 
 
class AlphaSpendingTest:
    def __init__(self, alpha: float = 0.05, method: str = "obrien_fleming"):
        self.alpha = alpha
        self.method = method
 
    def spending_function(self, tau: float) -> float:
        """정보 분율 tau까지 누적 소비할 alpha 계산"""
        tau = max(tau, 1e-10)  # 0 나눗셈 방지
        if self.method == "obrien_fleming":
            z_alpha = stats.norm.ppf(1 - self.alpha / 2)
            return 2 * (1 - stats.norm.cdf(z_alpha / np.sqrt(tau)))
        elif self.method == "pocock":
            return self.alpha * np.log(1 + (np.e - 1) * tau)
        else:
            raise ValueError(f"Unknown method: {self.method}")
 
    def get_critical_z(self, tau: float, prev_tau: float = 0.0) -> float:
        """
        현재 중간 분석 시점의 임계 Z-값 계산.
 
        prev_tau: 직전 분석 시점의 τ. 호출자가 이전 소비 이력을 관리해
        전달해야 한다. 기본값 0은 이번이 첫 번째 분석임을 의미한다.
        여러 번 호출할 때 매번 0으로 두면 누적 소비가 정확하게 추적되지
        않으므로, 상태 저장소(Redis 등)나 요청 파라미터로 prev_tau를 전달해야 한다.
        """
        # incremental_alpha: "이번 중간 분석에서 소비할 수 있는 남은 alpha 예산"
        incremental_alpha = (
            self.spending_function(tau) - self.spending_function(prev_tau)
        )
        # 수치 안정성: 경계값이 극단적으로 커지는 것 방지
        incremental_alpha = max(incremental_alpha, 1e-10)
        return stats.norm.ppf(1 - incremental_alpha / 2)
 
    def evaluate(self, z_stat: float, tau: float, prev_tau: float = 0.0) -> dict:
        boundary = self.get_critical_z(tau, prev_tau)
        return {
            "stop": abs(z_stat) > boundary,
            "z_stat": round(z_stat, 4),
            "boundary": round(boundary, 4),
            "tau": round(tau, 4),
            "incremental_alpha": round(
                self.spending_function(tau) - self.spending_function(prev_tau), 6
            ),
        }
 
 
def compute_z_stat(
    canary_error_rate: float,
    baseline_error_rate: float,
    n_canary: int,
    n_baseline: int,
) -> float:
    """
    이표본 비율 차이에 대한 Z-통계량.
 
    풀링된 표준오차(pooled SE)를 사용하는 이유: 귀무가설(두 비율이 같다) 아래서는
    pooled SE가 통계적으로 더 효율적이다. 대립가설이 사실일 때는 비풀링 SE가
    더 적절할 수 있지만, 회귀 감지가 목적인 Canary 배포에서는 귀무가설 기준
    검정이 표준 관행이다.
    """
    p_pool = (
        canary_error_rate * n_canary + baseline_error_rate * n_baseline
    ) / (n_canary + n_baseline)
    se = np.sqrt(p_pool * (1 - p_pool) * (1 / n_canary + 1 / n_baseline))
    if se < 1e-10:
        return 0.0
    return (canary_error_rate - baseline_error_rate) / se

# seq_test_service.py — FastAPI 엔드포인트 및 메트릭 수집
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
import os
 
app = FastAPI()
PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://prometheus:9090")
 
 
async def fetch_canary_metrics(canary_name: str) -> dict:
    """
    Prometheus에서 canary 메트릭 수집.
 
    주의: 아래 쿼리의 레이블(app="{canary_name}")은 예시다.
    실제 서비스의 레이블 구조(app, deployment, service 등)에 맞게 수정해야 한다.
    primary deployment는 "{canary_name}-primary" 레이블을 쓴다고 가정하지만
    실제 Flagger 환경에서는 다를 수 있다.
    """
    async with httpx.AsyncClient() as client:
        error_rate_query = (
            f'sum(rate(http_requests_total{{app="{canary_name}",'
            f'status=~"5.."}}[1m])) / '
            f'sum(rate(http_requests_total{{app="{canary_name}"}}[1m]))'
        )
        sample_count_query = (
            f'sum(increase(http_requests_total{{app="{canary_name}"}}[1h]))'
        )
 
        resp_error = await client.get(
            f"{PROMETHEUS_URL}/api/v1/query",
            params={"query": error_rate_query},
        )
        resp_count = await client.get(
            f"{PROMETHEUS_URL}/api/v1/query",
            params={"query": sample_count_query},
        )
 
        error_data = resp_error.json()["data"]["result"]
        count_data = resp_count.json()["data"]["result"]
 
        # 쿼리 결과가 비어있으면 메트릭 미수집 상태로 처리
        if not error_data or not count_data:
            raise ValueError(f"No metrics found for '{canary_name}'. "
                             "레이블 구조를 확인하세요.")
 
        error_rate = float(error_data[0]["value"][1])
        sample_count = float(count_data[0]["value"][1])
 
        return {"error_rate": error_rate, "sample_count": sample_count}
 
 
class WebhookPayload(BaseModel):
    metadata: dict = {}
 
 
@app.post("/check")
async def sequential_test_gate(payload: WebhookPayload):
    meta = payload.metadata
    alpha = float(meta.get("alpha", "0.05"))
    method = meta.get("method", "obrien_fleming")
    # Flagger metadata는 문자열만 전달되므로 int 변환 필요
    target_samples = int(meta.get("target_samples", "10000"))
    canary_name = meta.get("canary_name", "unknown")
    min_tau = float(meta.get("min_tau", "0.05"))  # 최소 웜업 5%
 
    # prev_tau 상태 관리: 현재 구현은 요청 파라미터로 받는 구조.
    # 실제 운영에서는 Redis 등 외부 상태 저장소에서 실험별 prev_tau를 관리해야
    # 매 분석마다 증분 소비가 정확하게 누적된다.
    prev_tau = float(meta.get("prev_tau", "0.0"))
 
    try:
        canary_metrics = await fetch_canary_metrics(canary_name)
        baseline_metrics = await fetch_canary_metrics(f"{canary_name}-primary")
    except Exception as e:
        # 메트릭 수집 실패 시 통과 처리 (fail-open 정책)
        # 서비스 특성에 따라 fail-closed로 변경할 수 있음
        return {"status": "metrics_unavailable", "reason": str(e)}
 
    n_canary = int(canary_metrics["sample_count"])
    tau = n_canary / target_samples
 
    # 최소 웜업 기간 미충족 시 통과 (τ가 너무 작으면 경계값이 극단적으로 높아짐)
    if tau < min_tau:
        return {"status": "warming_up", "tau": tau, "min_tau": min_tau}
 
    z_stat = compute_z_stat(
        canary_metrics["error_rate"],
        baseline_metrics["error_rate"],
        n_canary,
        max(int(baseline_metrics["sample_count"]), 1),
    )
 
    tester = AlphaSpendingTest(alpha=alpha, method=method)
    result = tester.evaluate(z_stat, min(tau, 1.0), prev_tau)
 
    if result["stop"]:
        raise HTTPException(
            status_code=400,
            detail={
                "reason": "sequential_test_boundary_exceeded",
                **result,
            },
        )
 
    return {"status": "pass", **result}

# visualize_boundaries.py — 로컬 분석 전용, 서비스에 포함하지 않는다
import matplotlib.pyplot as plt
import numpy as np
from scipy import stats
 
 
def obf_boundary(taus, alpha=0.05):
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    boundaries, prev_spend = [], 0
    for tau in taus:
        curr_spend = 2 * (1 - stats.norm.cdf(z_alpha / np.sqrt(tau)))
        inc = max(curr_spend - prev_spend, 1e-10)
        boundaries.append(stats.norm.ppf(1 - inc / 2))
        prev_spend = curr_spend
    return boundaries
 
 
def pocock_boundary(taus, alpha=0.05):
    boundaries, prev_spend = [], 0
    for tau in taus:
        curr_spend = alpha * np.log(1 + (np.e - 1) * tau)
        inc = max(curr_spend - prev_spend, 1e-10)
        boundaries.append(stats.norm.ppf(1 - inc / 2))
        prev_spend = curr_spend
    return boundaries
 
 
taus = np.linspace(0.01, 1.0, 100)
obf = obf_boundary(taus)
poc = pocock_boundary(taus)
 
plt.figure(figsize=(10, 5))
plt.plot(taus, obf, label="O'Brien-Fleming", linewidth=2)
plt.plot(taus, poc, label="Pocock", linewidth=2, linestyle="--")
plt.axhline(y=1.96, color="gray", linestyle=":", label="단일 검정 Z=1.96")
plt.xlabel("정보 분율 τ")
plt.ylabel("임계 Z-값")
plt.title("Alpha Spending 경계 비교 (target_samples=10000 기준)")
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig("boundary_comparison.png", dpi=150, bbox_inches="tight")