wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(agent99): route and dedupe alerts before dispatch

Browse Source
This commit is contained in:
ogt
2026-07-10 16:44:01 +08:00
parent ea98503a9a
commit 5c620844fe
7 changed files with 937 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
apps/api/src/api/v1/webhooks.py
View File

@@ -2850,22 +2850,6 @@ async def alertmanager_webhook(
labels=dict(alert.labels) if alert.labels else {},
annotations=dict(alert.annotations) if alert.annotations else {},
)
background_tasks.add_task(
bridge_alertmanager_to_agent99,
alert_id=alert_id,
alertname=alertname,
severity=severity,
namespace=namespace,
target_resource=target_resource,
message=message,
labels=dict(alert.labels) if alert.labels else {},
annotations=dict(alert.annotations) if alert.annotations else {},
fingerprint=fingerprint,
alert_category=alert_category,
notification_type=notification_type,
source_url=alert.generatorURL,
)
# ==========================================================================
# ADR-076: 告警聚合引擎 — 5 分鐘滑動視窗，防止告警風暴
# 2026-04-14 Claude Haiku 4.5 Asia/Taipei
@@ -2920,6 +2904,25 @@ async def alertmanager_webhook(
converged=True,
)
# Agent99 only receives the parent route after grouping. The bridge adds a
# Redis NX fingerprint lock so concurrent duplicate deliveries cannot race
# past this point and launch duplicate remediation.
background_tasks.add_task(
bridge_alertmanager_to_agent99,
alert_id=alert_id,
alertname=alertname,
severity=severity,
namespace=namespace,
target_resource=target_resource,
message=message,
labels=dict(alert.labels) if alert.labels else {},
annotations=dict(alert.annotations) if alert.annotations else {},
fingerprint=fingerprint,
alert_category=alert_category,
notification_type=notification_type,
source_url=alert.generatorURL,
)
try:
service = get_approval_service()

228
apps/api/src/services/agent99_sre_bridge.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import hashlib
import json
import re
import urllib.error
@@ -16,6 +17,8 @@ from src.utils.timezone import now_taipei
logger = get_logger("awoooi.agent99_sre_bridge")
AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS = 600
_SENSITIVE_KEY_RE = re.compile(
r"(token|secret|password|passwd|authorization|cookie|session|private[_-]?key)",
re.IGNORECASE,
@@ -99,18 +102,49 @@ def resolve_agent99_alert_kind(
text,
):
return "provider_freshness_signal"
if re.search(
r"wazuh|siem|security|intrusion|malware|auth_failed|bruteforce|"
r"privilege|vulnerability|\bcve\b|\bioc\b|suspicious|unauthorized",
text,
):
return "security_health"
if re.search(r"backup|snapshot|restore drill|restore_drill|backup cron", text):
return "backup_health"
if re.search(
r"postgres|postgresql|mysql|mariadb|database|db_connection|"
r"pg_isready|connection pool|replication lag",
text,
):
return "database_health"
if re.search(
r"alertchain|alert_chain|alert chain|alertmanager.*(broken|down|failed)|"
r"monitoring pipeline|notification pipeline",
text,
):
return "alert_chain_health"
if re.search(
r"reboot-auto-recovery|reboot_auto_recovery|reboot recovery|"
r"host_down|hostdown|host_reboot|vm_not_running|powered_off|"
r"all_required_hosts_not_in_10_minute_reboot_window",
text,
):
return "host_recovery"
if re.search(r"502|site_502|public_route|website|route_down|nginx upstream", text):
return "public_route_502"
if re.search(r"cpu|load|memory|mem_|disk|pressure|performance", text):
return "performance_pressure"
if re.search(r"backup|snapshot|restore drill|cron", text):
return "backup_health"
if re.search(r"harbor|registry", text):
return "harbor_registry"
if re.search(r"k3s|imagepullbackoff|errimagepull|crashloopbackoff|pod|deployment", text):
if re.search(
r"k3s|imagepullbackoff|errimagepull|crashloopbackoff|pod|deployment", text
):
return "awoooi_k3s"
if re.search(r"host_down|hostdown|vm_not_running|reboot|restart|powered_off", text):
return "host_recovery"
if re.search(
r"missing_key|e_ai_missing_key|application_ai_config_error|"
r"locale guard|ai support",
text,
):
return "application_ai_config_error"
return "monitoring_alert"
@@ -120,12 +154,95 @@ def _suggested_mode_for_kind(kind: str) -> str | None:
"public_route_502": "Recover",
"performance_pressure": "Perf",
"backup_health": "BackupCheck",
"security_health": "SecurityTriage",
"database_health": "Status",
"alert_chain_health": "Status",
"harbor_registry": "HarborRepair",
"awoooi_k3s": "AwoooRepair",
"host_recovery": "Recover",
"application_ai_config_error": "Status",
}.get(kind)
def _agent99_route_group_key(
*,
kind: str,
target_resource: str,
host: str,
) -> str:
parts = (
kind.strip().lower() or "monitoring_alert",
target_resource.strip().lower() or "unknown",
host.strip().lower() or "no-host",
)
return ":".join(_safe_file_token(part) for part in parts)
def agent99_sre_single_flight_key(fingerprint: str) -> str:
digest = hashlib.sha256(fingerprint.encode("utf-8")).hexdigest()
return f"agent99:sre_dispatch:single_flight:{digest}"
async def try_acquire_agent99_sre_single_flight(
fingerprint: str,
alert_id: str,
*,
ttl_seconds: int = AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS,
) -> bool:
"""Grant one Agent99 dispatch per fingerprint inside the controlled window."""
if not fingerprint:
return True
try:
from src.core.redis_client import get_redis
redis = get_redis()
acquired = await redis.set(
agent99_sre_single_flight_key(fingerprint),
alert_id,
ex=ttl_seconds,
nx=True,
)
return bool(acquired)
except Exception as exc:
logger.warning(
"agent99_sre_single_flight_failed_fail_open",
fingerprint=fingerprint,
alert_id=alert_id,
error=str(exc),
)
return True
async def release_agent99_sre_single_flight(fingerprint: str, alert_id: str) -> None:
"""Release only the caller-owned lock when transport did not accept the alert."""
if not fingerprint:
return
try:
from src.core.redis_client import get_redis
redis = get_redis()
await redis.eval(
"""
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
end
return 0
""",
1,
agent99_sre_single_flight_key(fingerprint),
alert_id,
)
except Exception as exc:
logger.warning(
"agent99_sre_single_flight_release_failed",
fingerprint=fingerprint,
alert_id=alert_id,
error=str(exc),
)
def build_agent99_sre_alert(
*,
alert_id: str,
@@ -164,6 +281,11 @@ def build_agent99_sre_alert(
or target_resource
or "awoooi"
)
route_group_key = _agent99_route_group_key(
kind=kind,
target_resource=target_resource,
host=host,
)
safe_labels = [
"alertmanager",
"awoooi-api",
@@ -172,9 +294,13 @@ def build_agent99_sre_alert(
f"target={_safe_label_value('target', target_resource)}",
]
if fingerprint:
safe_labels.append(f"fingerprint={_safe_label_value('fingerprint', fingerprint)}")
safe_labels.append(
f"fingerprint={_safe_label_value('fingerprint', fingerprint)}"
)
for key, value in sorted(labels.items()):
safe_labels.append(f"{_safe_text(key, max_length=80)}={_safe_label_value(key, value)}")
safe_labels.append(
f"{_safe_text(key, max_length=80)}={_safe_label_value(key, value)}"
)
if kind == "provider_freshness_signal":
safe_labels.extend(["provider-freshness", "no-false-green"])
@@ -197,8 +323,26 @@ def build_agent99_sre_alert(
"message": _safe_text(message, max_length=2000),
"labels": safe_labels,
"force": True,
"controlledApply": True,
"controlledApply": suggested_mode
in {
"Recover",
"StartVMs",
"HarborRepair",
"AwoooRepair",
"Perf",
"LoadShed",
},
"createdAt": now_taipei().isoformat(),
"routing": {
"schemaVersion": "agent99_alert_route_v1",
"kind": kind,
"suggestedMode": suggested_mode or "Status",
"routeSource": "structured_kind",
"groupKey": route_group_key,
"correlationKey": route_group_key,
"sourceFingerprint": _safe_text(fingerprint, max_length=160),
"singleFlightWindowSeconds": AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS,
},
"awoooi": {
"alertId": _safe_text(alert_id, max_length=160),
"alertname": _safe_text(alertname, max_length=160),
@@ -222,7 +366,9 @@ def write_agent99_sre_alert(
*,
inbox_path: str | None = None,
) -> Path | None:
target_dir_text = inbox_path if inbox_path is not None else settings.AGENT99_SRE_ALERT_INBOX_PATH
target_dir_text = (
inbox_path if inbox_path is not None else settings.AGENT99_SRE_ALERT_INBOX_PATH
)
if not target_dir_text:
logger.info(
"agent99_sre_bridge_disabled",
@@ -258,11 +404,17 @@ def post_agent99_sre_alert(
relay_token: str | None = None,
timeout_seconds: float | None = None,
) -> dict[str, Any] | None:
target_url = relay_url if relay_url is not None else settings.AGENT99_SRE_ALERT_RELAY_URL
target_url = (
relay_url if relay_url is not None else settings.AGENT99_SRE_ALERT_RELAY_URL
)
if not target_url:
return None
token = relay_token if relay_token is not None else settings.AGENT99_SRE_ALERT_RELAY_TOKEN
token = (
relay_token
if relay_token is not None
else settings.AGENT99_SRE_ALERT_RELAY_TOKEN
)
timeout = (
timeout_seconds
if timeout_seconds is not None
@@ -276,7 +428,9 @@ def post_agent99_sre_alert(
if token:
headers["X-Agent99-Relay-Token"] = token
request = urllib.request.Request(target_url, data=data, headers=headers, method="POST")
request = urllib.request.Request(
target_url, data=data, headers=headers, method="POST"
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
status = int(response.getcode())
@@ -331,7 +485,9 @@ async def bridge_alertmanager_to_agent99(
alert_category: str = "",
notification_type: str = "",
source_url: str | None = None,
) -> None:
) -> dict[str, Any]:
single_flight_fingerprint = ""
single_flight_acquired = False
try:
payload = build_agent99_sre_alert(
alert_id=alert_id,
@@ -347,11 +503,55 @@ async def bridge_alertmanager_to_agent99(
notification_type=notification_type,
source_url=source_url,
)
await asyncio.to_thread(dispatch_agent99_sre_alert, payload)
routing = (
payload.get("routing") if isinstance(payload.get("routing"), dict) else {}
)
single_flight_fingerprint = fingerprint or str(
routing.get("correlationKey") or ""
)
single_flight_acquired = await try_acquire_agent99_sre_single_flight(
single_flight_fingerprint,
alert_id,
)
if not single_flight_acquired:
logger.info(
"agent99_sre_dispatch_single_flight_suppressed",
alert_id=alert_id,
alertname=alertname,
fingerprint=single_flight_fingerprint,
group_key=routing.get("groupKey"),
)
return {
"status": "suppressed",
"reason": "single_flight_duplicate",
"fingerprint": single_flight_fingerprint,
"groupKey": routing.get("groupKey"),
}
dispatch_result = await asyncio.to_thread(dispatch_agent99_sre_alert, payload)
if dispatch_result == "disabled":
await release_agent99_sre_single_flight(single_flight_fingerprint, alert_id)
single_flight_acquired = False
return {
"status": "failed",
"reason": "bridge_disabled",
"fingerprint": single_flight_fingerprint,
"groupKey": routing.get("groupKey"),
}
return {
"status": "dispatched",
"dispatch": dispatch_result,
"fingerprint": single_flight_fingerprint,
"groupKey": routing.get("groupKey"),
"kind": payload.get("kind"),
"suggestedMode": payload.get("suggestedMode"),
}
except Exception as exc:
if single_flight_acquired:
await release_agent99_sre_single_flight(single_flight_fingerprint, alert_id)
logger.warning(
"agent99_sre_bridge_failed_fail_open",
alert_id=alert_id,
alertname=alertname,
error=str(exc),
)
return {"status": "failed", "reason": type(exc).__name__}