diff --git a/apps/api/src/core/circuit_breaker.py b/apps/api/src/core/circuit_breaker.py new file mode 100644 index 000000000..26e20754f --- /dev/null +++ b/apps/api/src/core/circuit_breaker.py @@ -0,0 +1,134 @@ +""" +OpenClaw 推理引擎保護機制 +========================= +ADR-038: 雙層保護策略 +- Layer 1: Circuit Breaker(防失敗傳播) +- Layer 2: Concurrency Semaphore(防 Thundering Herd) + +遵循 leWOOOgo 積木化鐵律: +- 此模組屬於 core/ 基礎設施層 +- 不依賴任何 Service 層 +- 透過 Singleton 提供全域狀態 + +2026-03-29 ogt - ADR-038 實作 +""" + +import asyncio +import time +from dataclasses import dataclass +from enum import Enum + +import structlog + +logger = structlog.get_logger(__name__) + + +class CircuitState(Enum): + CLOSED = "closed" # 正常運作 + OPEN = "open" # 斷路(快速失敗) + HALF_OPEN = "half_open" # 試探性恢復 + + +@dataclass +class CircuitBreakerConfig: + failure_threshold: int = 5 # 連續失敗次數觸發斷路 + timeout_s: float = 60.0 # 斷路後冷卻時間(秒) + max_concurrent: int = 3 # 最大並發 LLM 推理數 + + +class OpenClawGuard: + """ + OpenClaw 雙層推理保護門衛 + + 使用方式: + guard = get_openclaw_guard() + + if guard.is_circuit_open(): + return None # 快速失敗 + + async with guard.semaphore: # 排隊等待 + try: + result = await call_openclaw(...) + guard.record_success() + return result + except Exception: + guard.record_failure() + raise + """ + + def __init__(self, config: CircuitBreakerConfig | None = None): + self.config = config or CircuitBreakerConfig() + self.state = CircuitState.CLOSED + self.failure_count = 0 + self._opened_at: float | None = None + # Semaphore 必須在 event loop 中建立 + self._semaphore: asyncio.Semaphore | None = None + + @property + def semaphore(self) -> asyncio.Semaphore: + if self._semaphore is None: + self._semaphore = asyncio.Semaphore(self.config.max_concurrent) + return self._semaphore + + def is_circuit_open(self) -> bool: + """檢查 Circuit Breaker 是否處於斷路狀態""" + if self.state == CircuitState.OPEN: + if self._opened_at and time.time() - self._opened_at > self.config.timeout_s: + self.state = CircuitState.HALF_OPEN + logger.info("circuit_breaker_half_open") + return False + return True + return False + + def record_success(self) -> None: + """記錄成功呼叫,重置失敗計數""" + self.failure_count = 0 + if self.state != CircuitState.CLOSED: + logger.info("circuit_breaker_closed") + self.state = CircuitState.CLOSED + + def record_failure(self) -> None: + """記錄失敗呼叫,可能觸發斷路""" + self.failure_count += 1 + if self.failure_count >= self.config.failure_threshold: + self.state = CircuitState.OPEN + self._opened_at = time.time() + logger.warning( + "circuit_breaker_opened", + failure_count=self.failure_count, + cooldown_s=self.config.timeout_s, + ) + + def get_metrics(self) -> dict: + """取得目前狀態指標""" + return { + "state": self.state.value, + "failure_count": self.failure_count, + "max_concurrent": self.config.max_concurrent, + "opened_at": self._opened_at, + } + + def reset(self) -> None: + """重置狀態(供測試用)""" + self.state = CircuitState.CLOSED + self.failure_count = 0 + self._opened_at = None + self._semaphore = None + + +# 全域 Singleton +_guard: OpenClawGuard | None = None + + +def get_openclaw_guard() -> OpenClawGuard: + """取得全域 OpenClaw 保護門衛""" + global _guard + if _guard is None: + _guard = OpenClawGuard() + return _guard + + +def reset_openclaw_guard() -> None: + """重置全域 Guard(供測試用)""" + global _guard + _guard = None diff --git a/apps/api/src/services/auto_repair_service.py b/apps/api/src/services/auto_repair_service.py index 8943ce39e..f32f2c7e4 100644 --- a/apps/api/src/services/auto_repair_service.py +++ b/apps/api/src/services/auto_repair_service.py @@ -36,6 +36,10 @@ from src.models.playbook import ( ) from src.services.anomaly_counter import AnomalyFrequency, get_anomaly_counter from src.services.executor import get_executor +from src.services.global_repair_cooldown import ( + check_global_repair_cooldown, + record_global_repair_action, +) from src.services.playbook_service import IPlaybookService, get_playbook_service logger = structlog.get_logger(__name__) @@ -155,6 +159,23 @@ class AutoRepairService: severity=incident.severity.value if incident.severity else None, ) + # 0. 全域熔斷檢查(ADR-039 最優先) + can_repair, cooldown_reason = await check_global_repair_cooldown( + incident_id=incident.incident_id, + affected_services=incident.affected_services or [], + ) + if not can_repair: + logger.warning( + "auto_repair_blocked_global_cooldown", + incident_id=incident.incident_id, + reason=cooldown_reason, + ) + return AutoRepairDecision( + can_auto_repair=False, + reason=cooldown_reason, + blocked_by="GLOBAL_GUARDRAIL", + ) + # 1. 檢查 Incident 嚴重度 if incident.severity and incident.severity.value in ["P0", "P1"]: logger.info( @@ -262,6 +283,9 @@ class AutoRepairService: steps_count=len(playbook.repair_steps), ) + # ADR-039: 記錄全域修復計數(用於熔斷檢查) + await record_global_repair_action() + try: # 執行每個步驟 for step in playbook.repair_steps: diff --git a/apps/api/src/services/global_repair_cooldown.py b/apps/api/src/services/global_repair_cooldown.py new file mode 100644 index 000000000..020173960 --- /dev/null +++ b/apps/api/src/services/global_repair_cooldown.py @@ -0,0 +1,177 @@ +""" +全域修復熔斷機制 +================ +ADR-039:防止跨資源循環修復 + +設計原則: +- Redis TTL 滑動窗口(15 分鐘) +- 失敗降級:Redis 故障時保守跳過自動修復,強制人工確認 + +遵循 leWOOOgo 積木化鐵律: +- 此模組屬於 Service 層 +- 依賴 core/redis_client +- 無副作用的純邏輯判斷 + +2026-03-29 ogt - ADR-039 實作 +""" + +import structlog + +from src.core.redis_client import get_redis + +logger = structlog.get_logger(__name__) + +GLOBAL_COOLDOWN_KEY = "global:auto_repair:count" +GLOBAL_COOLDOWN_TTL = 900 # 15 分鐘窗口 +GLOBAL_COOLDOWN_THRESHOLD = 5 # 超過 5 次強制凍結 + +# ADR-039: 有狀態服務黑名單(永遠禁止自動重啟) +STATEFUL_SERVICE_BLACKLIST = frozenset( + { + # PostgreSQL + "postgres", + "postgresql", + "awoooi-postgres", + # Redis + "redis", + "awoooi-redis", + "redis-stack", + # ClickHouse (SignOz) + "clickhouse", + "signoz-clickhouse", + # 其他有狀態服務 + "elasticsearch", + "etcd", + "minio", + "awoooi-minio", + "kafka", + "zookeeper", + } +) + + +async def check_global_repair_cooldown( + incident_id: str, + affected_services: list[str] | None = None, +) -> tuple[bool, str]: + """ + 檢查是否允許自動修復 + + Args: + incident_id: 事件 ID(用於日誌追蹤) + affected_services: 受影響的服務列表 + + Returns: + (can_repair: bool, reason: str) + """ + affected_services = affected_services or [] + redis = get_redis() + + # === 硬禁令:有狀態服務黑名單 === + for service in affected_services: + service_lower = service.lower() + for blacklisted in STATEFUL_SERVICE_BLACKLIST: + if blacklisted in service_lower: + reason = f"服務 {service} 為有狀態服務,禁止自動重啟,請統帥手動介入" + logger.warning( + "stateful_service_blocked", + service=service, + incident_id=incident_id, + blacklist_match=blacklisted, + ) + return False, reason + + # === 全域冷卻期:Redis 計數 === + try: + count_raw = await redis.get(GLOBAL_COOLDOWN_KEY) + current_count = int(count_raw) if count_raw else 0 + + if current_count >= GLOBAL_COOLDOWN_THRESHOLD: + reason = ( + f"系統在過去 15 分鐘內已自動修復 {current_count} 次," + f"超出安全閾值 {GLOBAL_COOLDOWN_THRESHOLD}," + "強制轉為人工審核模式" + ) + logger.warning( + "global_repair_cooldown_active", + current_count=current_count, + threshold=GLOBAL_COOLDOWN_THRESHOLD, + incident_id=incident_id, + ) + return False, reason + + return True, "允許自動修復" + + except Exception as e: + # Redis 故障 → 保守策略:禁止自動修復 + logger.error( + "global_repair_cooldown_redis_error", + error=str(e), + fallback="blocking_auto_repair_for_safety", + incident_id=incident_id, + ) + return False, f"Redis 連線異常,保守禁止自動修復(原因:{e})" + + +async def record_global_repair_action() -> None: + """ + 記錄一次全域修復動作 + + 使用 INCR + EXPIRE 實現滑動窗口計數 + 注意:INCR 是原子操作,多個 Worker 並發安全 + """ + try: + redis = get_redis() + count = await redis.incr(GLOBAL_COOLDOWN_KEY) + + # 只在第一次設定 TTL(避免頻繁重設導致窗口延長) + if count == 1: + await redis.expire(GLOBAL_COOLDOWN_KEY, GLOBAL_COOLDOWN_TTL) + + logger.info( + "global_repair_action_recorded", + count=count, + threshold=GLOBAL_COOLDOWN_THRESHOLD, + ttl_seconds=GLOBAL_COOLDOWN_TTL, + ) + + except Exception as e: + # Redis 故障:靜默失敗(不影響主流程) + logger.warning("global_repair_record_failed", error=str(e)) + + +async def get_global_repair_status() -> dict: + """ + 取得全域修復狀態(供 API/監控用) + + Returns: + { + "current_count": int, + "threshold": int, + "is_frozen": bool, + "ttl_remaining": int | None, + } + """ + try: + redis = get_redis() + count_raw = await redis.get(GLOBAL_COOLDOWN_KEY) + current_count = int(count_raw) if count_raw else 0 + ttl = await redis.ttl(GLOBAL_COOLDOWN_KEY) + + return { + "current_count": current_count, + "threshold": GLOBAL_COOLDOWN_THRESHOLD, + "is_frozen": current_count >= GLOBAL_COOLDOWN_THRESHOLD, + "ttl_remaining": ttl if ttl > 0 else None, + "window_seconds": GLOBAL_COOLDOWN_TTL, + } + + except Exception as e: + logger.warning("get_global_repair_status_failed", error=str(e)) + return { + "current_count": -1, + "threshold": GLOBAL_COOLDOWN_THRESHOLD, + "is_frozen": True, # 保守假設:凍結 + "ttl_remaining": None, + "error": str(e), + } diff --git a/apps/api/tests/test_circuit_breaker.py b/apps/api/tests/test_circuit_breaker.py new file mode 100644 index 000000000..9efe86e5d --- /dev/null +++ b/apps/api/tests/test_circuit_breaker.py @@ -0,0 +1,164 @@ +""" +Circuit Breaker 測試 +==================== +ADR-038: OpenClaw 雙層保護機制 + +測試項目: +- Circuit Breaker 狀態轉換 +- Semaphore 並發控制 +- Graceful Degradation +""" + +import asyncio + +import pytest + +from src.core.circuit_breaker import ( + CircuitBreakerConfig, + CircuitState, + OpenClawGuard, + get_openclaw_guard, + reset_openclaw_guard, +) + + +class TestCircuitBreaker: + """Circuit Breaker 核心功能測試""" + + def setup_method(self): + """每個測試前重置全域 Guard""" + reset_openclaw_guard() + + def test_initial_state_is_closed(self): + """初始狀態應該是 CLOSED""" + guard = OpenClawGuard() + assert guard.state == CircuitState.CLOSED + assert guard.failure_count == 0 + assert not guard.is_circuit_open() + + def test_record_success_resets_failure_count(self): + """成功應該重置失敗計數""" + guard = OpenClawGuard() + guard.failure_count = 3 + guard.record_success() + assert guard.failure_count == 0 + + def test_record_failure_increments_count(self): + """失敗應該增加計數""" + guard = OpenClawGuard() + guard.record_failure() + assert guard.failure_count == 1 + guard.record_failure() + assert guard.failure_count == 2 + + def test_circuit_opens_after_threshold(self): + """連續失敗達到閾值後應該觸發斷路""" + config = CircuitBreakerConfig(failure_threshold=3) + guard = OpenClawGuard(config) + + guard.record_failure() + guard.record_failure() + assert guard.state == CircuitState.CLOSED + + guard.record_failure() + assert guard.state == CircuitState.OPEN + assert guard.is_circuit_open() + + def test_circuit_half_open_after_timeout(self): + """冷卻期後應該切換到 HALF_OPEN""" + config = CircuitBreakerConfig(failure_threshold=2, timeout_s=0.1) + guard = OpenClawGuard(config) + + # 觸發斷路 + guard.record_failure() + guard.record_failure() + assert guard.state == CircuitState.OPEN + assert guard.is_circuit_open() + + # 等待冷卻 + import time + + time.sleep(0.15) + + # 應該切換到 HALF_OPEN + assert not guard.is_circuit_open() + assert guard.state == CircuitState.HALF_OPEN + + def test_circuit_closes_after_success_in_half_open(self): + """HALF_OPEN 狀態下成功應該恢復 CLOSED""" + guard = OpenClawGuard() + guard.state = CircuitState.HALF_OPEN + + guard.record_success() + assert guard.state == CircuitState.CLOSED + assert guard.failure_count == 0 + + def test_get_metrics(self): + """應該正確返回指標""" + guard = OpenClawGuard() + guard.record_failure() + guard.record_failure() + + metrics = guard.get_metrics() + assert metrics["state"] == "closed" + assert metrics["failure_count"] == 2 + assert metrics["max_concurrent"] == 3 + + def test_singleton_pattern(self): + """全域 Guard 應該是單例""" + guard1 = get_openclaw_guard() + guard2 = get_openclaw_guard() + assert guard1 is guard2 + + def test_reset_clears_singleton(self): + """reset 應該清除單例""" + guard1 = get_openclaw_guard() + reset_openclaw_guard() + guard2 = get_openclaw_guard() + assert guard1 is not guard2 + + +class TestSemaphore: + """Semaphore 並發控制測試""" + + def setup_method(self): + reset_openclaw_guard() + + @pytest.mark.asyncio + async def test_semaphore_limits_concurrency(self): + """Semaphore 應該限制並發數""" + config = CircuitBreakerConfig(max_concurrent=2) + guard = OpenClawGuard(config) + + concurrent_count = 0 + max_concurrent_seen = 0 + results = [] + + async def worker(worker_id: int): + nonlocal concurrent_count, max_concurrent_seen + + async with guard.semaphore: + concurrent_count += 1 + max_concurrent_seen = max(max_concurrent_seen, concurrent_count) + await asyncio.sleep(0.05) + results.append(worker_id) + concurrent_count -= 1 + + # 啟動 5 個並發任務 + tasks = [asyncio.create_task(worker(i)) for i in range(5)] + await asyncio.gather(*tasks) + + # 確認最大並發數不超過限制 + assert max_concurrent_seen <= 2 + assert len(results) == 5 + + @pytest.mark.asyncio + async def test_semaphore_created_lazily(self): + """Semaphore 應該延遲建立""" + guard = OpenClawGuard() + assert guard._semaphore is None + + # 存取 semaphore 屬性會觸發建立 + sem = guard.semaphore + assert guard._semaphore is not None + assert sem is guard._semaphore diff --git a/apps/api/tests/test_global_repair_cooldown.py b/apps/api/tests/test_global_repair_cooldown.py new file mode 100644 index 000000000..307daec53 --- /dev/null +++ b/apps/api/tests/test_global_repair_cooldown.py @@ -0,0 +1,166 @@ +""" +Global Repair Cooldown 測試 +=========================== +ADR-039: 全域修復熔斷機制 + +測試項目: +- 有狀態服務黑名單檢查 +- 全域計數閾值 +- Redis 故障降級 + +注意:需要 Redis 環境,測試會使用獨立的 key 前綴 +""" + +import pytest + +from src.services.global_repair_cooldown import ( + GLOBAL_COOLDOWN_KEY, + GLOBAL_COOLDOWN_THRESHOLD, + STATEFUL_SERVICE_BLACKLIST, + check_global_repair_cooldown, + get_global_repair_status, + record_global_repair_action, +) + + +class TestStatefulServiceBlacklist: + """有狀態服務黑名單測試""" + + @pytest.mark.asyncio + async def test_postgres_blocked(self): + """PostgreSQL 服務應該被阻擋""" + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-001", + affected_services=["awoooi-postgres"], + ) + assert not can_repair + assert "有狀態服務" in reason + assert "禁止自動重啟" in reason + + @pytest.mark.asyncio + async def test_redis_blocked(self): + """Redis 服務應該被阻擋""" + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-002", + affected_services=["redis-stack"], + ) + assert not can_repair + assert "有狀態服務" in reason + + @pytest.mark.asyncio + async def test_clickhouse_blocked(self): + """ClickHouse 服務應該被阻擋""" + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-003", + affected_services=["signoz-clickhouse-0"], + ) + assert not can_repair + assert "有狀態服務" in reason + + @pytest.mark.asyncio + async def test_stateless_service_allowed(self): + """無狀態服務應該被允許""" + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-004", + affected_services=["awoooi-api-deployment"], + ) + assert can_repair + assert "允許" in reason + + @pytest.mark.asyncio + async def test_empty_services_allowed(self): + """空服務列表應該被允許""" + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-005", + affected_services=[], + ) + assert can_repair + + @pytest.mark.asyncio + async def test_none_services_allowed(self): + """None 服務列表應該被允許""" + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-006", + affected_services=None, + ) + assert can_repair + + def test_blacklist_contains_common_stateful_services(self): + """黑名單應該包含常見有狀態服務""" + assert "postgres" in STATEFUL_SERVICE_BLACKLIST + assert "redis" in STATEFUL_SERVICE_BLACKLIST + assert "clickhouse" in STATEFUL_SERVICE_BLACKLIST + assert "elasticsearch" in STATEFUL_SERVICE_BLACKLIST + assert "etcd" in STATEFUL_SERVICE_BLACKLIST + assert "minio" in STATEFUL_SERVICE_BLACKLIST + + +class TestGlobalCooldown: + """全域冷卻期測試 - 需要 Redis""" + + @pytest.fixture + async def clean_redis(self): + """清理測試用 Redis key""" + from src.core.redis_client import get_redis + + redis = get_redis() + await redis.delete(GLOBAL_COOLDOWN_KEY) + yield + await redis.delete(GLOBAL_COOLDOWN_KEY) + + @pytest.mark.asyncio + async def test_record_increments_counter(self, clean_redis): + """記錄應該增加計數""" + from src.core.redis_client import get_redis + + redis = get_redis() + + # 記錄一次 + await record_global_repair_action() + + count = await redis.get(GLOBAL_COOLDOWN_KEY) + assert int(count) == 1 + + @pytest.mark.asyncio + async def test_record_sets_ttl(self, clean_redis): + """第一次記錄應該設定 TTL""" + from src.core.redis_client import get_redis + + redis = get_redis() + + await record_global_repair_action() + + ttl = await redis.ttl(GLOBAL_COOLDOWN_KEY) + assert ttl > 0 + assert ttl <= 900 # 15 分鐘 + + @pytest.mark.asyncio + async def test_cooldown_triggers_after_threshold(self, clean_redis): + """超過閾值後應該觸發凍結""" + # 記錄 5 次(達到閾值) + for _ in range(GLOBAL_COOLDOWN_THRESHOLD): + await record_global_repair_action() + + can_repair, reason = await check_global_repair_cooldown( + incident_id="test-threshold", + affected_services=["awoooi-api"], + ) + + assert not can_repair + assert "超出安全閾值" in reason + assert str(GLOBAL_COOLDOWN_THRESHOLD) in reason + + @pytest.mark.asyncio + async def test_get_status_returns_correct_info(self, clean_redis): + """狀態 API 應該返回正確資訊""" + # 記錄 2 次 + await record_global_repair_action() + await record_global_repair_action() + + status = await get_global_repair_status() + + assert status["current_count"] == 2 + assert status["threshold"] == GLOBAL_COOLDOWN_THRESHOLD + assert not status["is_frozen"] + assert status["ttl_remaining"] is not None + assert status["ttl_remaining"] > 0