feat: add rate-channel breaker config and health_aware default
Threshold auto-raise now uses per-source concurrency only (the M2 global-concurrency formula neutered the breaker at scale). Four new optional keys: MIN_CALLS, FAIL_RATE, WINDOW_S, MAX_COOLDOWN_S.
This commit is contained in:
@@ -45,7 +45,7 @@ _SOURCE_FIELDS: dict[str, tuple[str, str]] = {
|
|||||||
"TRUST_ENV": ("trust_env", "bool"),
|
"TRUST_ENV": ("trust_env", "bool"),
|
||||||
}
|
}
|
||||||
_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"})
|
_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"})
|
||||||
_SELECTORS = frozenset({"round_robin", "least_inflight"})
|
_SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"})
|
||||||
_QUOTA_FULL = frozenset({"wait", "fail_fast"})
|
_QUOTA_FULL = frozenset({"wait", "fail_fast"})
|
||||||
# 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源)
|
# 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源)
|
||||||
_DEFAULT_STALL_WINDOW_S = 300.0
|
_DEFAULT_STALL_WINDOW_S = 300.0
|
||||||
@@ -136,7 +136,7 @@ class GatewaySettings:
|
|||||||
retry=retry,
|
retry=retry,
|
||||||
breaker=breaker,
|
breaker=breaker,
|
||||||
backpressure=_load_backpressure(scope_u, env),
|
backpressure=_load_backpressure(scope_u, env),
|
||||||
selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "round_robin"),
|
selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "health_aware"),
|
||||||
quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"),
|
quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"),
|
||||||
**_load_pgw(env),
|
**_load_pgw(env),
|
||||||
)
|
)
|
||||||
@@ -211,10 +211,9 @@ def _load_breaker(
|
|||||||
key_c, cool = _require(env, f"{scope}__BREAKER__COOLDOWN_S", "LLM_CIRCUIT_BREAKER_COOLDOWN")
|
key_c, cool = _require(env, f"{scope}__BREAKER__COOLDOWN_S", "LLM_CIRCUIT_BREAKER_COOLDOWN")
|
||||||
threshold = int(_cast(thr, "int", key_t))
|
threshold = int(_cast(thr, "int", key_t))
|
||||||
cooldown_s = float(_cast(cool, "float", key_c))
|
cooldown_s = float(_cast(cool, "float", key_c))
|
||||||
# 有效阈值 = max(配置值, 并发×2)——三项目 .env 注释的手动约定入库(设计 §9 行 4)
|
# 有效阈值 = max(配置值, 源级并发×2)。M2.5 修正: 只看源级并发——M2 曾用
|
||||||
concurrency = global_limits.max_concurrency or max(
|
# 全局并发抬升(SOAK 100→阈值 200)使熔断失灵(病灶 2,设计 2026-07-21-m25)
|
||||||
(s.max_concurrency for s in sources), default=0
|
concurrency = max((s.max_concurrency for s in sources), default=0)
|
||||||
)
|
|
||||||
if concurrency > 0:
|
if concurrency > 0:
|
||||||
threshold = max(threshold, concurrency * 2)
|
threshold = max(threshold, concurrency * 2)
|
||||||
slowest = max(s.timeout_s for s in sources)
|
slowest = max(s.timeout_s for s in sources)
|
||||||
@@ -231,7 +230,28 @@ def _load_breaker(
|
|||||||
else:
|
else:
|
||||||
# 派生规则: 探针租约须撑过一次最慢调用,且不短于冷却期(第三项保证守卫恒成立)
|
# 派生规则: 探针租约须撑过一次最慢调用,且不短于冷却期(第三项保证守卫恒成立)
|
||||||
probe_ttl_s = max(2 * slowest, cooldown_s, probe_floor)
|
probe_ttl_s = max(2 * slowest, cooldown_s, probe_floor)
|
||||||
return BreakerConfig(fail_threshold=threshold, cooldown_s=cooldown_s, probe_ttl_s=probe_ttl_s)
|
# M2.5 失败率通道参数(可选键,库缺省——韧性参数缺省先例同 backpressure)
|
||||||
|
min_calls = int(_opt_float(env, f"{scope}__BREAKER__MIN_CALLS", 10))
|
||||||
|
fail_rate = _opt_float(env, f"{scope}__BREAKER__FAIL_RATE", 0.6)
|
||||||
|
window_s = _opt_float(env, f"{scope}__BREAKER__WINDOW_S", 60.0)
|
||||||
|
max_cooldown_s = _opt_float(env, f"{scope}__BREAKER__MAX_COOLDOWN_S", max(300.0, cooldown_s))
|
||||||
|
return BreakerConfig(
|
||||||
|
fail_threshold=threshold,
|
||||||
|
cooldown_s=cooldown_s,
|
||||||
|
probe_ttl_s=probe_ttl_s,
|
||||||
|
min_calls=min_calls,
|
||||||
|
fail_rate=fail_rate,
|
||||||
|
window_s=window_s,
|
||||||
|
max_cooldown_s=max_cooldown_s,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _opt_float(env: Mapping[str, str], key: str, default: float) -> float:
|
||||||
|
"""可选韧性参数: 缺省用库值,显式配置则解析(坏值 fail-loud)。"""
|
||||||
|
raw = env.get(key)
|
||||||
|
if raw is None or raw == "":
|
||||||
|
return default
|
||||||
|
return float(_cast(raw, "float", key))
|
||||||
|
|
||||||
|
|
||||||
def _load_backpressure(scope: str, env: Mapping[str, str]) -> BackpressurePolicy:
|
def _load_backpressure(scope: str, env: Mapping[str, str]) -> BackpressurePolicy:
|
||||||
|
|||||||
@@ -152,15 +152,28 @@ class RetryPolicy:
|
|||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class BreakerConfig:
|
class BreakerConfig:
|
||||||
"""熔断配置;probe_ttl_s 是半开探针租约时长(持有者死亡后自动回收)。"""
|
"""熔断配置;probe_ttl_s 是半开探针租约时长(持有者死亡后自动回收)。
|
||||||
|
|
||||||
|
M2.5 双通道: fail_threshold 是连续失败通道;min_calls/fail_rate/window_s
|
||||||
|
是失败率通道(窗口样本 ≥ min_calls 且失败率 ≥ fail_rate 即开路,429 不入);
|
||||||
|
开路时长按重开次数指数递增,封顶 max_cooldown_s(设计 2026-07-21-m25)。
|
||||||
|
"""
|
||||||
|
|
||||||
fail_threshold: int
|
fail_threshold: int
|
||||||
cooldown_s: float
|
cooldown_s: float
|
||||||
probe_ttl_s: float
|
probe_ttl_s: float
|
||||||
|
min_calls: int = 10
|
||||||
|
fail_rate: float = 0.6
|
||||||
|
window_s: float = 60.0
|
||||||
|
max_cooldown_s: float = 300.0
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
def __post_init__(self) -> None:
|
||||||
if self.fail_threshold < 1 or self.cooldown_s <= 0 or self.probe_ttl_s <= 0:
|
if self.fail_threshold < 1 or self.cooldown_s <= 0 or self.probe_ttl_s <= 0:
|
||||||
raise ValueError("熔断配置要求 fail_threshold ≥ 1 且 cooldown_s/probe_ttl_s > 0")
|
raise ValueError("熔断配置要求 fail_threshold ≥ 1 且 cooldown_s/probe_ttl_s > 0")
|
||||||
|
if self.min_calls < 1 or not (0.0 < self.fail_rate <= 1.0) or self.window_s <= 0:
|
||||||
|
raise ValueError("失败率通道要求 min_calls ≥ 1、0 < fail_rate ≤ 1、window_s > 0")
|
||||||
|
if self.max_cooldown_s < self.cooldown_s:
|
||||||
|
raise ValueError("max_cooldown_s 不得小于 cooldown_s(退避封顶低于初值)")
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
|
|||||||
@@ -126,8 +126,15 @@ class TestResilienceKeys:
|
|||||||
GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"}))
|
GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"}))
|
||||||
|
|
||||||
def test_selector_and_quota_full(self):
|
def test_selector_and_quota_full(self):
|
||||||
|
# M2.5: 缺省选源改 health_aware(生产级默认);显式配置者不变
|
||||||
s = GatewaySettings.from_env("LLM", env=_env())
|
s = GatewaySettings.from_env("LLM", env=_env())
|
||||||
assert s.selector == "round_robin" and s.quota_full == "wait"
|
assert s.selector == "health_aware" and s.quota_full == "wait"
|
||||||
|
assert (
|
||||||
|
GatewaySettings.from_env(
|
||||||
|
"LLM", env=_env(**{"LLM__SELECTOR": "round_robin"})
|
||||||
|
).selector
|
||||||
|
== "round_robin"
|
||||||
|
)
|
||||||
s2 = GatewaySettings.from_env(
|
s2 = GatewaySettings.from_env(
|
||||||
"LLM", env=_env(**{"LLM__SELECTOR": "least_inflight", "LLM__QUOTA_FULL": "fail_fast"})
|
"LLM", env=_env(**{"LLM__SELECTOR": "least_inflight", "LLM__QUOTA_FULL": "fail_fast"})
|
||||||
)
|
)
|
||||||
@@ -172,9 +179,42 @@ class TestAssemblyGuards:
|
|||||||
def test_effective_breaker_threshold_auto_raised(self):
|
def test_effective_breaker_threshold_auto_raised(self):
|
||||||
env = _env(**{"LLM__QWEN__1__MAX_CONCURRENCY": "8"})
|
env = _env(**{"LLM__QWEN__1__MAX_CONCURRENCY": "8"})
|
||||||
s = GatewaySettings.from_env("LLM", env=env)
|
s = GatewaySettings.from_env("LLM", env=env)
|
||||||
# 有效阈值 = max(配置值 5, 并发 8 × 2) = 16(.env 注释约定入库)
|
# 有效阈值 = max(配置值 5, 源级并发 8 × 2) = 16(M2.5: 抬升只看源级)
|
||||||
assert s.breaker.fail_threshold == 16
|
assert s.breaker.fail_threshold == 16
|
||||||
|
|
||||||
|
def test_breaker_threshold_not_raised_by_global_concurrency(self):
|
||||||
|
# M2.5 病灶 2 回归: 全局并发不再抬升阈值(M2 曾 max(5, 100×2)=200 使熔断失灵)
|
||||||
|
env = _env(**{"LLM__GLOBAL__MAX_CONCURRENCY": "100", "LLM__GLOBAL__RPM": "600"})
|
||||||
|
s = GatewaySettings.from_env("LLM", env=env)
|
||||||
|
assert s.breaker.fail_threshold == 5
|
||||||
|
|
||||||
|
def test_breaker_rate_channel_defaults_and_overrides(self):
|
||||||
|
# M2.5 失败率通道参数: 库缺省(韧性参数缺省先例)与显式覆盖
|
||||||
|
s = GatewaySettings.from_env("LLM", env=_env())
|
||||||
|
assert s.breaker.min_calls == 10
|
||||||
|
assert s.breaker.fail_rate == pytest.approx(0.6)
|
||||||
|
assert s.breaker.window_s == pytest.approx(60.0)
|
||||||
|
assert s.breaker.max_cooldown_s == pytest.approx(300.0) # max(300, cooldown 60)
|
||||||
|
env = _env(
|
||||||
|
**{
|
||||||
|
"LLM__BREAKER__MIN_CALLS": "20",
|
||||||
|
"LLM__BREAKER__FAIL_RATE": "0.5",
|
||||||
|
"LLM__BREAKER__WINDOW_S": "30",
|
||||||
|
"LLM__BREAKER__MAX_COOLDOWN_S": "600",
|
||||||
|
"LLM_CIRCUIT_BREAKER_COOLDOWN": "400",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
s2 = GatewaySettings.from_env("LLM", env=env)
|
||||||
|
assert s2.breaker.min_calls == 20 and s2.breaker.fail_rate == pytest.approx(0.5)
|
||||||
|
assert s2.breaker.window_s == pytest.approx(30.0)
|
||||||
|
assert s2.breaker.max_cooldown_s == pytest.approx(600.0)
|
||||||
|
|
||||||
|
def test_breaker_max_cooldown_floor_follows_cooldown(self):
|
||||||
|
# 缺省封顶 = max(300, cooldown): 冷却 400s 时封顶随之 400s
|
||||||
|
env = _env(**{"LLM_CIRCUIT_BREAKER_COOLDOWN": "400"})
|
||||||
|
s = GatewaySettings.from_env("LLM", env=env)
|
||||||
|
assert s.breaker.max_cooldown_s == pytest.approx(400.0)
|
||||||
|
|
||||||
def test_redis_governance_backend_requires_url(self):
|
def test_redis_governance_backend_requires_url(self):
|
||||||
"""M2 解禁 redis 后端: 取 redis 时 REDIS_URL 必在,缺则装配报错。"""
|
"""M2 解禁 redis 后端: 取 redis 时 REDIS_URL 必在,缺则装配报错。"""
|
||||||
env = _env(
|
env = _env(
|
||||||
|
|||||||
Reference in New Issue
Block a user