fix(verification): align playbook and mcp evidence for canary alerts
All checks were successful
Code Review / ai-code-review (push) Successful in 18s
CD Pipeline / tests (push) Successful in 1m2s
CD Pipeline / build-and-deploy (push) Successful in 3m31s
CD Pipeline / post-deploy-checks (push) Successful in 1m39s

This commit is contained in:
Your Name
2026-05-14 00:21:39 +08:00
parent 42d0d076d6
commit b1ecb55bd6
3 changed files with 22 additions and 5 deletions

View File

@@ -170,6 +170,11 @@ class MCPToolRegistry:
list[RegisteredTool]: 推薦工具(已排序)
"""
suggested: list[RegisteredTool] = []
labels = incident_labels or {}
has_k8s_locator = any(
labels.get(key)
for key in ("deployment", "pod", "node", "namespace", "container")
)
# 依優先度排序後篩選
sorted_tools = sorted(self._tools, key=lambda t: t.priority)
@@ -181,7 +186,9 @@ class MCPToolRegistry:
# incident_type_hints 過濾
if reg.incident_type_hints:
if not any(alertname.startswith(hint) for hint in reg.incident_type_hints):
matches_hint = any(alertname.startswith(hint) for hint in reg.incident_type_hints)
is_k8s_state_tool = SensorDimension.D1_K8S_STATE in reg.dimensions
if not matches_hint and not (has_k8s_locator and is_k8s_state_tool):
continue
# 感官維度去重(每個維度取優先度最高的一個工具即可)

View File

@@ -96,10 +96,11 @@ async def _resolve_exact_yaml_rule(
"""
SELECT playbook_id
FROM playbooks
WHERE source = 'yaml_rule'
AND status = 'approved'
WHERE status = 'approved'
AND (symptom_pattern::jsonb->'alert_names') ? :alertname
ORDER BY updated_at DESC
ORDER BY
CASE WHEN source = 'yaml_rule' THEN 0 ELSE 1 END,
updated_at DESC
LIMIT 1
"""
),

View File

@@ -21,7 +21,6 @@ import pytest
from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
from src.services.mcp_tool_registry import (
MCPToolRegistry,
RegisteredTool,
SensorDimension,
_classify_tool,
)
@@ -267,6 +266,16 @@ class TestSuggestTools:
assert "kubectl_describe" not in names
assert "prometheus_query" in names
def test_custom_alert_with_deployment_label_gets_k8s_tool(self):
registry = self._registry_with_tools()
tools = registry.suggest_tools(
alertname="AwoooPT16Canary",
incident_labels={"deployment": "awoooi-auto-repair-canary", "namespace": "awoooi-prod"},
)
names = [t.tool.name for t in tools]
assert "kubectl_describe" in names
assert "prometheus_query" in names
def test_empty_alertname_gets_generic_only(self):
registry = self._registry_with_tools()
tools = registry.suggest_tools(alertname="")