feat(sweeper): 新增 Incident Analysis Sweeper — 自動觸發未分析 Incident AI 決策
Gap修復: Signal Worker 創建 Incident 後,AI 分析只在 GET /api/v1/incidents 被呼叫時觸發 若前端無人訪問,新 Incident 永遠沒有 AI 分析與 Telegram 通知 解法: 新增 src/jobs/incident_analysis_sweeper.py 每 90 秒掃描無 decision token 的 INVESTIGATING incidents 自動背景觸發 get_or_create_decision() — Semaphore(3) 限流,每批最多 5 筆 main.py lifespan 啟動時 asyncio.create_task() 掛載 2026-04-16 Claude Sonnet 4.6 Asia/Taipei Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
106
apps/api/src/jobs/incident_analysis_sweeper.py
Normal file
106
apps/api/src/jobs/incident_analysis_sweeper.py
Normal file
@@ -0,0 +1,106 @@
|
||||
"""
|
||||
Incident Analysis Sweeper — 自動觸發 INVESTIGATING 事件 AI 分析
|
||||
================================================================
|
||||
問題背景:
|
||||
Signal Worker 創建 Incident 後,AI 分析 (decision_manager) 原本只在
|
||||
GET /api/v1/incidents 被呼叫時才觸發 (背景 fire-and-forget)。
|
||||
若前端沒人看或 Telegram Bot 未呼叫該端點,新 Incident 永遠沒有 AI 分析。
|
||||
|
||||
解法:
|
||||
每 90 秒掃描 INVESTIGATING 狀態且無 decision token 的 Incident,
|
||||
自動在背景觸發 get_or_create_decision()。
|
||||
|
||||
限流:
|
||||
Semaphore(3) — 避免並發壓垮 OPENCLAW_NEMO/Ollama
|
||||
每批最多 5 個 incident,避免啟動雪崩
|
||||
|
||||
2026-04-16 Claude Sonnet 4.6 Asia/Taipei
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
import structlog
|
||||
|
||||
from src.models.incident import Incident, IncidentStatus, Severity
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
_SWEEP_INTERVAL_SEC = 90 # 每 90 秒掃一次
|
||||
_MAX_BATCH = 5 # 每批最多 5 個
|
||||
_SEMAPHORE_LIMIT = 3 # 最多 3 個並發 AI 分析
|
||||
|
||||
|
||||
async def run_incident_analysis_sweeper() -> None:
|
||||
"""
|
||||
永久迴圈:每 90 秒自動為未分析的 INVESTIGATING Incident 觸發 AI 分析。
|
||||
由 main.py lifespan 透過 asyncio.create_task() 啟動。
|
||||
"""
|
||||
logger.info("incident_analysis_sweeper_started", interval_sec=_SWEEP_INTERVAL_SEC)
|
||||
sem = asyncio.Semaphore(_SEMAPHORE_LIMIT)
|
||||
|
||||
while True:
|
||||
try:
|
||||
await _sweep_once(sem)
|
||||
except Exception as e:
|
||||
logger.warning("incident_analysis_sweeper_error", error=str(e))
|
||||
|
||||
await asyncio.sleep(_SWEEP_INTERVAL_SEC)
|
||||
|
||||
|
||||
async def _sweep_once(sem: asyncio.Semaphore) -> None:
|
||||
"""
|
||||
執行一次掃描:找出沒有 decision token 的 INVESTIGATING incidents,
|
||||
在背景觸發 AI 分析。
|
||||
"""
|
||||
from src.services.decision_manager import get_decision_manager
|
||||
from src.services.incident_service import get_incident_service
|
||||
from src.core.redis_client import get_redis
|
||||
|
||||
redis = get_redis()
|
||||
incident_service = get_incident_service()
|
||||
dm = get_decision_manager()
|
||||
|
||||
# 取得所有 INVESTIGATING incidents
|
||||
try:
|
||||
incidents: list[Incident] = await incident_service.get_active_incidents()
|
||||
except Exception as e:
|
||||
logger.warning("sweeper_get_incidents_failed", error=str(e))
|
||||
return
|
||||
|
||||
if not incidents:
|
||||
return
|
||||
|
||||
# 找出沒有 decision token 的
|
||||
unanalyzed = []
|
||||
for incident in incidents:
|
||||
token_key = f"decision:{incident.incident_id}"
|
||||
if not await redis.exists(token_key):
|
||||
unanalyzed.append(incident)
|
||||
|
||||
if not unanalyzed:
|
||||
return
|
||||
|
||||
# 限制每批
|
||||
batch = unanalyzed[:_MAX_BATCH]
|
||||
logger.info(
|
||||
"sweeper_triggering_analysis",
|
||||
total_unanalyzed=len(unanalyzed),
|
||||
batch_size=len(batch),
|
||||
)
|
||||
|
||||
async def _analyze(incident: Incident) -> None:
|
||||
async with sem:
|
||||
try:
|
||||
timeout = 120.0 if incident.severity in (Severity.P0, Severity.P1) else 180.0
|
||||
await dm.get_or_create_decision(incident=incident, timeout_sec=timeout)
|
||||
logger.info("sweeper_analysis_done", incident_id=incident.incident_id)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"sweeper_analysis_failed",
|
||||
incident_id=incident.incident_id,
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
tasks = [asyncio.create_task(_analyze(inc)) for inc in batch]
|
||||
await asyncio.gather(*tasks, return_exceptions=True)
|
||||
@@ -334,6 +334,17 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
|
||||
except Exception as e:
|
||||
logger.warning("stale_ready_tokens_resend_schedule_failed", error=str(e))
|
||||
|
||||
# 2026-04-16 Claude Sonnet 4.6: 自動 AI 分析 Sweeper(每 90 秒)
|
||||
# 修復核心 Gap:Signal Worker 創建 Incident 後無人觸發 AI 分析
|
||||
# 除非有人呼叫 GET /api/v1/incidents,否則 Incident 永遠沒有決策
|
||||
# Sweeper 定期掃描無 decision token 的 INVESTIGATING incidents → 背景觸發
|
||||
try:
|
||||
from src.jobs.incident_analysis_sweeper import run_incident_analysis_sweeper
|
||||
asyncio.create_task(run_incident_analysis_sweeper())
|
||||
logger.info("incident_analysis_sweeper_scheduled", interval_sec=90)
|
||||
except Exception as e:
|
||||
logger.warning("incident_analysis_sweeper_schedule_failed", error=str(e))
|
||||
|
||||
# ADR-076 Task 4: 每日 08:00 台北時間自動日度巡檢報告
|
||||
# 2026-04-14 Claude Haiku 4.5 Asia/Taipei
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user