025a493f06
Wave 8 P3.2 模型版本追蹤 + ADR-100 SLO 自我治理 + 配套： P3.2 — Model Version Tracking: - model_version_probe.py (268 行) — 探測 Ollama / OpenRouter 等 provider 的 model version - model_version_tracker.py (101 行) — 對齊 PG provider_version_history 表 - migrations/p3_2_provider_version_history.sql + rollback — 25 行 schema - db/models.py +32 行 — ProviderVersionHistory ORM ADR-100 — AI 自主化 SLO: - docs/adr/ADR-100-ai-autonomous-slo.md (167 行) — 飛輪 SLO 設計與閾值 - ops/monitoring/slo-rules.yml (254 行) — Prometheus SLO recording rules + alerts - ops/monitoring/tests/test_slo_rules.yaml (242 行) — promtool unit tests 整合修改: - main.py +72 行 — Lifespan 啟動 model_version_probe + KB rot cleaner schedule - gitea_webhook.py +45 行 — webhook 接收 model 版本變化通知 - ci_auto_repair.py / evidence_snapshot.py / pre_decision_investigator.py — 配合接線 新測試: - test_kb_rot_cleaner_schedule.py (120 行) — 9 tests pass - test_slo_rules.yaml — promtool 驗收 Tests: 9 passed (test_kb_rot_cleaner_schedule) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Co-Authored-By: Multiple Engineers (P3.2 + ADR-100) <noreply@anthropic.com>
647 lines
26 KiB
Python
647 lines
26 KiB
Python
"""
|
||
AWOOOI AIOps Phase 1 — 決策前情報調查員
|
||
==========================================
|
||
在 LLM 做出任何決策之前，主動呼叫 MCP 工具蒐集 8D 感官情報，
|
||
並將結果封裝為不可變的 EvidenceSnapshot。
|
||
|
||
設計原則：
|
||
1. 工具動態選擇（不 hardcode）— 從 MCPToolRegistry.suggest_tools() 取清單
|
||
2. 並行蒐集（asyncio.gather）— 8D 感官同步展開，P99 < 8s
|
||
3. 部分失敗不阻塞（Graceful Degradation）— 某感官失敗標 mcp_health[tool]=False，繼續其他
|
||
4. Prompt Injection 防護（Sanitization）— 所有文字輸入先過 SanitizationService
|
||
5. Redis 快取（30s 滑動窗口）— 防告警風暴重複打 K8s API
|
||
|
||
快取 Key 格式：
|
||
evidence:{sha256(alertname + namespace + pod_name + severity)[:12]}
|
||
|
||
P99 延遲目標：< 8000ms（超時個別工具丟棄，不阻塞主路徑）
|
||
Token Budget：單次 evidence_summary ≤ 32,000 chars（≈ 8K tokens）
|
||
|
||
ADR-081: PreDecisionInvestigator + EvidenceSnapshot
|
||
MASTER §3.1.3 (A)(B)(C)
|
||
2026-04-15 ogt + Claude Sonnet 4.6 (亞太): Phase 1 初始建立
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import time
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
import structlog
|
||
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
from src.services.mcp_tool_registry import RegisteredTool, SensorDimension, get_mcp_tool_registry
|
||
from src.services.sanitization_service import sanitize, sanitize_dict_values
|
||
|
||
if TYPE_CHECKING:
|
||
from src.models.incident import Incident
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# 單一 MCP 工具呼叫的超時（秒）— 超過則丟棄，不阻塞主路徑
|
||
MCP_TOOL_TIMEOUT_SEC = 5.0
|
||
|
||
# 全局 Investigator 超時（P99 目標）
|
||
INVESTIGATOR_TIMEOUT_SEC = 8.0
|
||
|
||
# Redis 快取 TTL（秒）
|
||
CACHE_TTL_SEC = 30
|
||
|
||
|
||
class PreDecisionInvestigator:
|
||
"""
|
||
決策前情報調查員。
|
||
|
||
每個 Incident 在 LLM 推理前，先由此服務蒐集 8D 感官數據，
|
||
產出 EvidenceSnapshot 作為 LLM 的「眼睛」。
|
||
|
||
Usage:
|
||
investigator = PreDecisionInvestigator()
|
||
snapshot = await investigator.investigate(incident)
|
||
# snapshot.evidence_summary 可直接貼進 LLM prompt
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._registry = get_mcp_tool_registry()
|
||
|
||
async def investigate(self, incident: "Incident") -> EvidenceSnapshot:
|
||
"""
|
||
主入口：為 Incident 蒐集 8D 感官情報。
|
||
|
||
流程：
|
||
1. 計算 fingerprint → 查 Redis cache
|
||
2. cache miss → 並行呼叫 suggest_tools() 回傳的工具
|
||
3. 每個工具結果過 SanitizationService
|
||
4. 組裝 EvidenceSnapshot → 寫 incident_evidence 表
|
||
5. 寫 Redis cache
|
||
|
||
Args:
|
||
incident: 目前處理中的 Incident
|
||
|
||
Returns:
|
||
EvidenceSnapshot: 含 evidence_summary 的完整快照
|
||
（即使所有 MCP 失敗也回傳空快照，不 raise）
|
||
"""
|
||
start_ms = int(time.monotonic() * 1000)
|
||
incident_id = incident.incident_id if hasattr(incident, "incident_id") else str(incident.id)
|
||
|
||
# 1. 計算 fingerprint 並查 cache
|
||
fingerprint = _compute_fingerprint(incident)
|
||
cached = await _get_cache(fingerprint)
|
||
if cached is not None:
|
||
logger.debug("investigator_cache_hit", incident_id=incident_id, fingerprint=fingerprint)
|
||
return cached
|
||
|
||
# 2. 取工具清單
|
||
alertname = _get_alertname(incident)
|
||
labels = _get_labels(incident)
|
||
tools = self._registry.suggest_tools(
|
||
alertname=alertname,
|
||
incident_labels=labels,
|
||
)
|
||
|
||
snapshot = EvidenceSnapshot(incident_id=incident_id)
|
||
snapshot.sensors_attempted = len(tools)
|
||
|
||
# 告警基礎資訊：sensors=0 時 AI 至少知道是什麼告警
|
||
# 2026-04-16 ogt + Claude Sonnet 4.6: 修復空 evidence → ABSTAIN 問題
|
||
sigs = getattr(incident, "signals", []) or []
|
||
sig0 = sigs[0] if sigs else None
|
||
snapshot.alert_info = {
|
||
"alert_name": alertname or getattr(incident, "alertname", "") or "",
|
||
"severity": str(getattr(incident, "severity", "")),
|
||
"affected_services": getattr(incident, "affected_services", []) or [],
|
||
"labels": labels,
|
||
"annotations": (
|
||
({k: v for k, v in (sig0.annotations or {}).items()} if sig0 else {})
|
||
),
|
||
"source": getattr(sig0, "source", "") if sig0 else "",
|
||
"incident_id": incident_id,
|
||
}
|
||
|
||
# 3. 並行蒐集（整體 INVESTIGATOR_TIMEOUT_SEC 保護）
|
||
try:
|
||
await asyncio.wait_for(
|
||
self._collect_all(snapshot, tools, incident),
|
||
timeout=INVESTIGATOR_TIMEOUT_SEC,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.warning(
|
||
"investigator_global_timeout",
|
||
incident_id=incident_id,
|
||
timeout_sec=INVESTIGATOR_TIMEOUT_SEC,
|
||
)
|
||
|
||
# 4. 記錄耗時
|
||
snapshot.collection_duration_ms = int(time.monotonic() * 1000) - start_ms
|
||
|
||
# 4.5 Phase 4 8D 感官增強：填入動態異常上下文（非阻塞，失敗不影響主路徑）
|
||
try:
|
||
await asyncio.wait_for(
|
||
self._collect_phase4_anomalies(snapshot),
|
||
timeout=2.0, # Phase 4 感官最多等 2s，不能拖慢主路徑
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.warning("phase4_anomaly_collect_timeout", incident_id=incident_id)
|
||
except Exception:
|
||
logger.exception("phase4_anomaly_collect_error", incident_id=incident_id)
|
||
|
||
# 4.6 P3.1-T2 by Claude 2026-04-27 — DiagnosisAggregator Pod 深診斷（守門：ENABLE_DIAGNOSIS_AGGREGATOR）
|
||
# Conservative 策略：預設關閉，避免與 MCP sensor 重複收集 K8s+SignOz 資料。
|
||
# 待重疊分析完成確認互補性後，由統帥設定 ENABLE_DIAGNOSIS_AGGREGATOR=true 啟用。
|
||
try:
|
||
from src.core.config import settings as _settings
|
||
if _settings.ENABLE_DIAGNOSIS_AGGREGATOR:
|
||
await asyncio.wait_for(
|
||
self._collect_diagnosis_aggregator(snapshot, incident),
|
||
timeout=3.0,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.warning("diagnosis_aggregator_collect_timeout", incident_id=incident_id)
|
||
except Exception:
|
||
logger.warning("diagnosis_aggregator_collect_failed", incident_id=incident_id)
|
||
|
||
# 5. 組裝 summary
|
||
snapshot.evidence_summary = snapshot.build_summary()
|
||
|
||
# 6. 持久化（fire-and-await，Phase 3 學習閉環依賴此表）
|
||
try:
|
||
await snapshot.save()
|
||
except Exception:
|
||
logger.exception("investigator_save_failed", incident_id=incident_id)
|
||
# 不 raise：snapshot 仍可用於決策，存儲失敗不阻塞主路徑
|
||
|
||
# 7. 寫 cache
|
||
await _set_cache(fingerprint, snapshot)
|
||
|
||
logger.info(
|
||
"investigator_done",
|
||
incident_id=incident_id,
|
||
sensors_attempted=snapshot.sensors_attempted,
|
||
sensors_succeeded=snapshot.sensors_succeeded,
|
||
duration_ms=snapshot.collection_duration_ms,
|
||
)
|
||
return snapshot
|
||
|
||
async def _collect_diagnosis_aggregator(
|
||
self,
|
||
snapshot: EvidenceSnapshot,
|
||
incident: "Incident", # noqa: ARG002 — 路徑 A 從 snapshot 取 raw 資料，不需 incident labels
|
||
) -> None:
|
||
"""
|
||
2026-04-27 P3.1-T2-PathA by Claude — DiagAggregator 信號分類層補 PDI
|
||
|
||
路徑 A：用 DA 的信號分類補 PDI raw 資料。
|
||
不重複收集 K8s/SignOz，只取 raw 資料（來自 PDI 已收集的 D1/D2/D3）
|
||
丟給 DA.classify_signals_from_raw() 做業務邏輯分類（OOMKilled/CrashLoop/HighLatency 等）。
|
||
結果以結構化 dict 存入 snapshot.extra_diagnosis。
|
||
"""
|
||
from src.services.diagnosis_aggregator import get_diagnosis_aggregator
|
||
|
||
try:
|
||
aggregator = get_diagnosis_aggregator()
|
||
|
||
# 從 snapshot 取 PDI 已收集的 raw 資料（不打外部 API）
|
||
signals = aggregator.classify_signals_from_raw(
|
||
k8s_data=snapshot.k8s_state,
|
||
logs_data=snapshot.recent_logs,
|
||
metrics_data=snapshot.metrics_snapshot,
|
||
)
|
||
|
||
result = {
|
||
"signal_count": len(signals),
|
||
"signals": [s.to_dict() if hasattr(s, "to_dict") else str(s) for s in signals],
|
||
}
|
||
snapshot.extra_diagnosis = result
|
||
|
||
logger.debug(
|
||
"diagnosis_aggregator_signal_classify_done",
|
||
incident_id=snapshot.incident_id,
|
||
signal_count=len(signals),
|
||
)
|
||
except Exception as e:
|
||
logger.warning("diagnosis_aggregator_signal_classify_failed", error=str(e))
|
||
|
||
async def _collect_phase4_anomalies(self, snapshot: EvidenceSnapshot) -> None:
|
||
"""
|
||
Phase 4 8D 感官增強：從 ProactiveInspector 快取 + LogAnomalyDetector
|
||
讀取動態異常上下文，填入 snapshot.anomaly_context。
|
||
|
||
設計原則：
|
||
- 只讀快取，不觸發新的 Prometheus 查詢（避免延遲）
|
||
- 失敗靜默降級（外層已包 try/except + timeout）
|
||
- Phase 4 Shadow Mode 時資料仍填入（供 LLM 參考，不觸發 Alert）
|
||
|
||
2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 4 8D 升級
|
||
"""
|
||
from src.services.proactive_inspector import get_proactive_inspector
|
||
from src.services.log_anomaly_detector import get_log_anomaly_detector
|
||
|
||
context: dict[str, Any] = {}
|
||
|
||
# 1. 讀取最近一次巡檢報告（ProactiveInspector 每 5 分鐘更新一次）
|
||
inspector = get_proactive_inspector()
|
||
last_report = inspector.get_last_report()
|
||
if last_report is not None:
|
||
context["last_inspection_at"] = last_report.finished_at
|
||
context["shadow_mode"] = last_report.shadow_mode
|
||
|
||
if last_report.baseline_anomalies > 0:
|
||
context["baseline_anomalies"] = [
|
||
{
|
||
"metric": a.metric_name,
|
||
"severity": a.severity,
|
||
"description": a.description,
|
||
"deviation_sigma": a.deviation_sigma,
|
||
}
|
||
for a in last_report.alerts
|
||
if a.alert_type == "dynamic_anomaly"
|
||
]
|
||
|
||
if last_report.trend_breaches > 0:
|
||
context["trend_breaches"] = [
|
||
{
|
||
"metric": a.metric_name,
|
||
"description": a.description,
|
||
"breach_in_hours": a.predicted_breach_hours,
|
||
}
|
||
for a in last_report.alerts
|
||
if a.alert_type == "trend_breach"
|
||
]
|
||
|
||
# 2. 讀取最近新 log pattern（最多 5 個）
|
||
detector = get_log_anomaly_detector()
|
||
recent_patterns = await detector.get_recent_new_patterns(limit=5)
|
||
if recent_patterns:
|
||
context["recent_log_patterns"] = [
|
||
{
|
||
"template": p.get("template", "")[:200],
|
||
"cluster_id": p.get("cluster_id", ""),
|
||
"source": p.get("source", ""),
|
||
}
|
||
for p in recent_patterns
|
||
]
|
||
|
||
if context:
|
||
snapshot.anomaly_context = context
|
||
logger.debug(
|
||
"phase4_anomaly_context_collected",
|
||
has_baseline=bool(context.get("baseline_anomalies")),
|
||
has_trends=bool(context.get("trend_breaches")),
|
||
log_patterns=len(context.get("recent_log_patterns", [])),
|
||
)
|
||
|
||
async def _collect_all(
|
||
self,
|
||
snapshot: EvidenceSnapshot,
|
||
tools: list[RegisteredTool],
|
||
incident: "Incident",
|
||
) -> None:
|
||
"""並行呼叫所有工具，結果填入 snapshot。"""
|
||
params = _build_tool_params(incident)
|
||
|
||
tasks = [
|
||
self._collect_one(snapshot, reg, params)
|
||
for reg in tools
|
||
]
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
async def _collect_one(
|
||
self,
|
||
snapshot: EvidenceSnapshot,
|
||
reg: RegisteredTool,
|
||
params: dict[str, Any],
|
||
) -> None:
|
||
"""執行單一 MCP 工具呼叫，結果填入對應感官維度。"""
|
||
tool_name = reg.tool.name
|
||
snapshot.mcp_health[tool_name] = False # 預設失敗，成功後覆蓋
|
||
|
||
_started = asyncio.get_event_loop().time()
|
||
_mcp_status = "failed"
|
||
_mcp_error = None
|
||
try:
|
||
result = await asyncio.wait_for(
|
||
reg.provider.execute(tool_name, params),
|
||
timeout=MCP_TOOL_TIMEOUT_SEC,
|
||
)
|
||
|
||
if not result.success:
|
||
logger.warning(
|
||
"investigator_tool_failed",
|
||
tool=tool_name,
|
||
error=result.error,
|
||
)
|
||
_mcp_error = str(result.error)[:200] if result.error else "unknown"
|
||
return
|
||
|
||
snapshot.mcp_health[tool_name] = True
|
||
snapshot.sensors_succeeded += 1
|
||
_mcp_status = "success"
|
||
|
||
# 依感官維度填入對應欄位
|
||
raw = result.output
|
||
_fill_snapshot_dimension(snapshot, reg, raw)
|
||
|
||
except asyncio.TimeoutError:
|
||
logger.warning("investigator_tool_timeout", tool=tool_name, timeout=MCP_TOOL_TIMEOUT_SEC)
|
||
_mcp_status = "timeout"
|
||
_mcp_error = f"timeout {MCP_TOOL_TIMEOUT_SEC}s"
|
||
except Exception as _e:
|
||
logger.exception("investigator_tool_error", tool=tool_name)
|
||
_mcp_status = "error"
|
||
_mcp_error = str(_e)[:200]
|
||
finally:
|
||
# 2026-04-18 ADR-090-D: MCP 呼叫入 timeline_events(MASTER §7.1 #4 KPI)
|
||
try:
|
||
_duration_ms = int((asyncio.get_event_loop().time() - _started) * 1000)
|
||
asyncio.create_task(_log_mcp_call_to_timeline(
|
||
snapshot_incident_id=getattr(snapshot, "incident_id", None),
|
||
provider_name=reg.provider.name,
|
||
tool_name=tool_name,
|
||
status=_mcp_status,
|
||
error=_mcp_error,
|
||
duration_ms=_duration_ms,
|
||
))
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _log_mcp_call_to_timeline(
|
||
snapshot_incident_id: str | None,
|
||
provider_name: str,
|
||
tool_name: str,
|
||
status: str,
|
||
error: str | None,
|
||
duration_ms: int,
|
||
) -> None:
|
||
"""
|
||
2026-04-18 ADR-090-D: MCP 呼叫寫入 timeline_events,支援 MASTER §7.1 #4
|
||
"MCP 呼叫次數/24h > 0" KPI 量測。
|
||
"""
|
||
try:
|
||
from sqlalchemy import text as _sql
|
||
from src.db.base import get_db_context
|
||
import json as _json
|
||
_description = _json.dumps({
|
||
"provider": provider_name,
|
||
"tool": tool_name,
|
||
"status": status,
|
||
"error": error,
|
||
"duration_ms": duration_ms,
|
||
}, ensure_ascii=False)
|
||
async with get_db_context() as _db:
|
||
await _db.execute(
|
||
_sql("""
|
||
INSERT INTO timeline_events (
|
||
incident_id, event_type, status, title, description, actor,
|
||
actor_role, created_at
|
||
) VALUES (
|
||
:iid, 'mcp_call', :st, :tl, :desc, :actor,
|
||
'mcp', NOW()
|
||
)
|
||
"""),
|
||
{
|
||
"iid": snapshot_incident_id or "unknown",
|
||
"st": status,
|
||
"tl": f"MCP {provider_name}.{tool_name}"[:100],
|
||
"desc": _description[:500],
|
||
"actor": provider_name[:50],
|
||
},
|
||
)
|
||
except Exception:
|
||
# 靜默失敗,timeline_events 是稽核,不能反噬 MCP 主流程
|
||
pass
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Snapshot dimension mapping
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _fill_snapshot_dimension(
|
||
snapshot: EvidenceSnapshot,
|
||
reg: RegisteredTool,
|
||
raw: Any,
|
||
) -> None:
|
||
"""將工具輸出填入 EvidenceSnapshot 對應感官欄位。"""
|
||
if raw is None:
|
||
return
|
||
|
||
for dim in reg.dimensions:
|
||
if dim == SensorDimension.D1_K8S_STATE:
|
||
if isinstance(raw, dict):
|
||
snapshot.k8s_state = sanitize_dict_values(raw, "k8s_state")
|
||
else:
|
||
snapshot.k8s_state = {"raw": sanitize(str(raw), "k8s_state")}
|
||
|
||
elif dim == SensorDimension.D2_LOGS:
|
||
text = raw if isinstance(raw, str) else json.dumps(raw, ensure_ascii=False)
|
||
snapshot.recent_logs = sanitize(text, "recent_logs")
|
||
|
||
elif dim == SensorDimension.D3_METRICS:
|
||
if isinstance(raw, dict):
|
||
snapshot.metrics_snapshot = sanitize_dict_values(raw, "metrics")
|
||
else:
|
||
snapshot.metrics_snapshot = {"raw": str(raw)}
|
||
|
||
elif dim == SensorDimension.D4_CHANGES:
|
||
# Gate 1 fix: 過 sanitize_dict_values，ArgoCD diff / Git commit message 可含注入
|
||
if isinstance(raw, list):
|
||
snapshot.recent_deployments = [
|
||
sanitize_dict_values(item, "d4_changes") if isinstance(item, dict)
|
||
else {"raw": sanitize(str(item), "d4_changes")}
|
||
for item in raw
|
||
]
|
||
elif isinstance(raw, dict):
|
||
snapshot.recent_deployments = [sanitize_dict_values(raw, "d4_changes")]
|
||
|
||
elif dim == SensorDimension.D5_BUSINESS:
|
||
# Gate 1 fix: 業務指標可能含 Grafana annotation 等外部字串
|
||
if isinstance(raw, dict):
|
||
snapshot.business_metrics = sanitize_dict_values(raw, "d5_business")
|
||
|
||
elif dim == SensorDimension.D6_HISTORY:
|
||
text = raw if isinstance(raw, str) else json.dumps(raw, ensure_ascii=False)
|
||
snapshot.historical_context = sanitize(text, "historical_context")[:2000]
|
||
|
||
elif dim == SensorDimension.D7_PEERS:
|
||
# Gate 1 fix: Pod annotation / label 可含注入
|
||
if isinstance(raw, dict):
|
||
snapshot.peer_health = sanitize_dict_values(raw, "d7_peers")
|
||
|
||
elif dim == SensorDimension.D8_TOPOLOGY:
|
||
# Gate 1 fix: Istio / service mesh metadata 可含外部字串
|
||
if isinstance(raw, dict):
|
||
snapshot.dependency_topology = sanitize_dict_values(raw, "d8_topology")
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Helpers
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _get_alertname(incident: "Incident") -> str:
|
||
if incident.signals:
|
||
sig = incident.signals[0]
|
||
# alert_name 在 Signal 頂層欄位，labels["alertname"] 是 Prometheus 慣例但可能為空
|
||
return (
|
||
getattr(sig, "alert_name", "")
|
||
or sig.labels.get("alertname", "")
|
||
or getattr(incident, "alertname", "")
|
||
or ""
|
||
)
|
||
return getattr(incident, "alertname", "") or ""
|
||
|
||
|
||
def _get_labels(incident: "Incident") -> dict[str, Any]:
|
||
if incident.signals:
|
||
sig = incident.signals[0]
|
||
labels = sig.labels or {}
|
||
# 若 labels 缺少 alertname，補上頂層的 alert_name
|
||
if "alertname" not in labels and getattr(sig, "alert_name", ""):
|
||
labels = dict(labels)
|
||
labels["alertname"] = sig.alert_name
|
||
return labels
|
||
return {}
|
||
|
||
|
||
_SHORT_HOST_MAP: dict[str, str] = {
|
||
"110": "192.168.0.110",
|
||
"120": "192.168.0.120",
|
||
"121": "192.168.0.121",
|
||
"188": "192.168.0.188",
|
||
}
|
||
"""
|
||
Prometheus instance label 使用短主機名（如 "110:9100"），
|
||
SSH_MCP_ALLOWED_HOSTS 使用完整 IP（如 "192.168.0.110"）。
|
||
此映射表做轉換，避免 SSH 工具 "Host 'X' not in SSH_MCP_ALLOWED_HOSTS" 失敗。
|
||
2026-04-16 ogt + Claude Sonnet 4.6: 修復 sensors 7/8 失敗根因
|
||
"""
|
||
|
||
|
||
def _build_prometheus_query(alertname: str, namespace: str, pod_name: str) -> str:
|
||
"""依告警類型生成 Prometheus PromQL 查詢（供 prometheus_query tool 使用）。
|
||
2026-04-24 ogt + Claude Sonnet 4.6: P0.4 fix — _build_tool_params 補 query 欄位"""
|
||
an = alertname.lower()
|
||
# CPU / 負載
|
||
if any(k in an for k in ("cpu", "load", "throttl")):
|
||
filter_pod = f',pod=~"{pod_name}.*"' if pod_name else ""
|
||
return f'avg(rate(container_cpu_usage_seconds_total{{namespace="{namespace}"{filter_pod}}}[5m]))'
|
||
# 記憶體
|
||
elif any(k in an for k in ("memory", "mem", "oom")):
|
||
filter_pod = f',pod=~"{pod_name}.*"' if pod_name else ""
|
||
return f'avg(container_memory_working_set_bytes{{namespace="{namespace}"{filter_pod}}}) / 1048576'
|
||
# CrashLoop / 重啟
|
||
elif any(k in an for k in ("crash", "restart", "oom", "backoff")):
|
||
return f'sum(increase(kube_pod_container_status_restarts_total{{namespace="{namespace}"}}[15m]))'
|
||
# 磁碟 / 儲存
|
||
elif any(k in an for k in ("disk", "storage", "pvc", "volume", "capacity")):
|
||
return 'sum(kubelet_volume_stats_used_bytes) by (persistentvolumeclaim)'
|
||
# HTTP / 可用性
|
||
elif any(k in an for k in ("http", "error", "5xx", "probe", "down", "unhealthy")):
|
||
return '1 - avg(probe_success)'
|
||
# Pod / Container 狀態
|
||
elif any(k in an for k in ("pod", "container", "deploy", "replicaset")):
|
||
return f'kube_pod_status_phase{{namespace="{namespace}"}}'
|
||
# 通用 fallback
|
||
else:
|
||
return f'up{{namespace="{namespace}"}}'
|
||
|
||
|
||
def _build_tool_params(incident: "Incident") -> dict[str, Any]:
|
||
"""從 Incident 提取 MCP 工具呼叫所需的公共參數。"""
|
||
labels = _get_labels(incident)
|
||
raw_host = labels.get("instance", "").split(":")[0] or labels.get("host", "")
|
||
host = _SHORT_HOST_MAP.get(raw_host, raw_host) # 短名 → 完整 IP
|
||
namespace = labels.get("namespace", "awoooi-prod")
|
||
pod_name = labels.get("pod", labels.get("name", ""))
|
||
alertname = labels.get("alertname", "")
|
||
return {
|
||
"namespace": namespace,
|
||
"pod_name": pod_name,
|
||
"deployment": labels.get("deployment", ""),
|
||
"host": host,
|
||
"container": labels.get("container", labels.get("name", "")),
|
||
"alertname": alertname,
|
||
# P0.4 fix 2026-04-24 ogt + Claude Sonnet 4.6: Prometheus tool 需要 query 欄位
|
||
# 原本缺少此欄位 → prometheus_query/range tool 傳入空 query → 回傳 error dict
|
||
"query": _build_prometheus_query(alertname, namespace, pod_name),
|
||
}
|
||
|
||
|
||
def _compute_fingerprint(incident: "Incident") -> str:
|
||
"""計算 cache key 用的 fingerprint。"""
|
||
labels = _get_labels(incident)
|
||
key = ":".join([
|
||
labels.get("alertname", ""),
|
||
labels.get("namespace", ""),
|
||
labels.get("pod", labels.get("name", "")),
|
||
labels.get("severity", ""),
|
||
])
|
||
return hashlib.sha256(key.encode()).hexdigest()[:16]
|
||
|
||
|
||
async def _get_cache(fingerprint: str) -> EvidenceSnapshot | None:
|
||
"""從 Redis 取快取的 EvidenceSnapshot（若存在）。"""
|
||
try:
|
||
from src.core.redis_client import get_redis
|
||
redis = get_redis()
|
||
key = f"evidence:{fingerprint}"
|
||
raw = await redis.get(key)
|
||
if raw is None:
|
||
return None
|
||
|
||
data = json.loads(raw)
|
||
snap = EvidenceSnapshot(
|
||
incident_id=data.get("incident_id", ""),
|
||
snapshot_id=data.get("snapshot_id", ""),
|
||
)
|
||
snap.evidence_summary = data.get("evidence_summary", "")
|
||
snap.k8s_state = data.get("k8s_state")
|
||
snap.recent_logs = data.get("recent_logs")
|
||
snap.metrics_snapshot = data.get("metrics_snapshot")
|
||
snap.mcp_health = data.get("mcp_health", {})
|
||
snap.sensors_attempted = data.get("sensors_attempted", 0)
|
||
snap.sensors_succeeded = data.get("sensors_succeeded", 0)
|
||
return snap
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
async def _set_cache(fingerprint: str, snapshot: EvidenceSnapshot) -> None:
|
||
"""將 EvidenceSnapshot 寫入 Redis cache。"""
|
||
try:
|
||
from src.core.redis_client import get_redis
|
||
redis = get_redis()
|
||
key = f"evidence:{fingerprint}"
|
||
payload = {
|
||
"incident_id": snapshot.incident_id,
|
||
"snapshot_id": snapshot.snapshot_id,
|
||
"evidence_summary": snapshot.evidence_summary,
|
||
"k8s_state": snapshot.k8s_state,
|
||
"recent_logs": snapshot.recent_logs,
|
||
"metrics_snapshot": snapshot.metrics_snapshot,
|
||
"mcp_health": snapshot.mcp_health,
|
||
"sensors_attempted": snapshot.sensors_attempted,
|
||
"sensors_succeeded": snapshot.sensors_succeeded,
|
||
}
|
||
await redis.set(key, json.dumps(payload, ensure_ascii=False), ex=CACHE_TTL_SEC)
|
||
except Exception:
|
||
pass # cache 失敗不影響主路徑
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Singleton
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_investigator: PreDecisionInvestigator | None = None
|
||
|
||
|
||
def get_pre_decision_investigator() -> PreDecisionInvestigator:
|
||
"""取得 PreDecisionInvestigator Singleton。"""
|
||
global _investigator
|
||
if _investigator is None:
|
||
_investigator = PreDecisionInvestigator()
|
||
return _investigator
|