30153496d1
- Import sorting (I001) - Unused imports (F401) - f-string without placeholders (F541) - Loop variable unused (B007) - zip() strict parameter (B905) - Exception chaining (B904) - collections.abc imports (UP035) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
319 lines
11 KiB
Python
319 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
🚀 AWOOOI Phase 2 導彈腳本 - fire_test_alert.py
|
|
===============================================
|
|
向系統注入模擬告警，觸發 OpenClaw AI 分析流程
|
|
|
|
用途:
|
|
- 驗證全鏈路 (Webhook → OpenClaw → ApprovalCard)
|
|
- 測試戰情室前端是否即時彈出授權卡片
|
|
- 開發除錯用 (無需真實監控系統)
|
|
|
|
執行方式:
|
|
cd apps/api
|
|
python -m scripts.fire_test_alert
|
|
|
|
# 指定告警類型
|
|
python -m scripts.fire_test_alert --type db_connection_timeout
|
|
python -m scripts.fire_test_alert --type k8s_pod_crash --severity critical
|
|
|
|
Author: Claude Code
|
|
Date: 2026-03-21
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import httpx
|
|
|
|
# =============================================================================
|
|
# Config
|
|
# =============================================================================
|
|
|
|
API_BASE_URL = "http://localhost:8000"
|
|
WEBHOOK_ENDPOINT = f"{API_BASE_URL}/api/v1/webhooks/alerts"
|
|
|
|
# =============================================================================
|
|
# 預定義告警場景 (High-Fidelity Mock Alerts)
|
|
# =============================================================================
|
|
|
|
ALERT_SCENARIOS = {
|
|
"db_connection_timeout": {
|
|
"alert_type": "db_connection_timeout",
|
|
"severity": "critical",
|
|
"source": "prometheus-alertmanager",
|
|
"target_resource": "postgres-primary-0",
|
|
"namespace": "database",
|
|
"message": "PostgreSQL Database OOM - Connection pool exhausted, 47 waiting queries",
|
|
"metrics": {
|
|
"connection_count": 100,
|
|
"waiting_queries": 47,
|
|
"cpu_percent": 89,
|
|
"memory_percent": 95,
|
|
"sigma_deviation": 4.2,
|
|
},
|
|
"labels": {
|
|
"app": "postgres",
|
|
"team": "dba",
|
|
"tier": "critical",
|
|
},
|
|
},
|
|
"k8s_pod_crash": {
|
|
"alert_type": "k8s_pod_crash",
|
|
"severity": "warning",
|
|
"source": "k8s-event-watcher",
|
|
"target_resource": "harbor-core-7d4b8c9f5-xk2m3",
|
|
"namespace": "harbor",
|
|
"message": "Pod CrashLoopBackOff detected - OOMKilled after 5 restarts",
|
|
"metrics": {
|
|
"restart_count": 5,
|
|
"last_exit_code": 137,
|
|
"cpu_percent": 95,
|
|
"memory_percent": 100,
|
|
"sigma_deviation": 3.8,
|
|
},
|
|
"labels": {
|
|
"app": "harbor-core",
|
|
"team": "devops",
|
|
},
|
|
},
|
|
"high_cpu": {
|
|
"alert_type": "high_cpu",
|
|
"severity": "warning",
|
|
"source": "node-exporter",
|
|
"target_resource": "api-backend-deployment",
|
|
"namespace": "production",
|
|
"message": "Payment API Latency Spike - CPU at 94%, response time > 2s",
|
|
"metrics": {
|
|
"cpu_percent": 94,
|
|
"memory_percent": 72,
|
|
"response_time_ms": 2340,
|
|
"sigma_deviation": 3.2,
|
|
},
|
|
"labels": {
|
|
"app": "payment-api",
|
|
"team": "backend",
|
|
"sla": "critical",
|
|
},
|
|
},
|
|
"disk_full": {
|
|
"alert_type": "disk_full",
|
|
"severity": "critical",
|
|
"source": "node-exporter",
|
|
"target_resource": "logging-node-01",
|
|
"namespace": "kube-system",
|
|
"message": "Disk usage at 97% - /var/log nearly full, risk of logging failure",
|
|
"metrics": {
|
|
"disk_percent": 97,
|
|
"available_gb": 2.3,
|
|
"inode_percent": 89,
|
|
},
|
|
"labels": {
|
|
"node": "logging-node-01",
|
|
"team": "sre",
|
|
},
|
|
},
|
|
"ssl_expiry": {
|
|
"alert_type": "ssl_expiry",
|
|
"severity": "warning",
|
|
"source": "cert-manager",
|
|
"target_resource": "awoooi.wooo.work",
|
|
"namespace": "cert-manager",
|
|
"message": "SSL Certificate expiring in 7 days - auto-renewal failed",
|
|
"metrics": {
|
|
"days_until_expiry": 7,
|
|
},
|
|
"labels": {
|
|
"domain": "awoooi.wooo.work",
|
|
"issuer": "letsencrypt",
|
|
},
|
|
},
|
|
}
|
|
|
|
# =============================================================================
|
|
# Terminal Output Helpers (漂亮的 Log)
|
|
# =============================================================================
|
|
|
|
class Colors:
|
|
"""ANSI Color Codes"""
|
|
HEADER = '\033[95m'
|
|
BLUE = '\033[94m'
|
|
CYAN = '\033[96m'
|
|
GREEN = '\033[92m'
|
|
YELLOW = '\033[93m'
|
|
RED = '\033[91m'
|
|
ENDC = '\033[0m'
|
|
BOLD = '\033[1m'
|
|
DIM = '\033[2m'
|
|
|
|
|
|
def print_banner():
|
|
"""Print AWOOOI ASCII Banner"""
|
|
banner = f"""
|
|
{Colors.CYAN}{Colors.BOLD}
|
|
█████╗ ██╗ ██╗ ██████╗ ██████╗ ██████╗ ██╗
|
|
██╔══██╗██║ ██║██╔═══██╗██╔═══██╗██╔═══██╗██║
|
|
███████║██║ █╗ ██║██║ ██║██║ ██║██║ ██║██║
|
|
██╔══██║██║███╗██║██║ ██║██║ ██║██║ ██║██║
|
|
██║ ██║╚███╔███╔╝╚██████╔╝╚██████╔╝╚██████╔╝██║
|
|
╚═╝ ╚═╝ ╚══╝╚══╝ ╚═════╝ ╚═════╝ ╚═════╝ ╚═╝
|
|
{Colors.ENDC}
|
|
{Colors.DIM} 🚀 Phase 2 導彈腳本 - Test Alert Injector{Colors.ENDC}
|
|
{Colors.DIM} ─────────────────────────────────────────{Colors.ENDC}
|
|
"""
|
|
print(banner)
|
|
|
|
|
|
def print_section(title: str):
|
|
"""Print section header"""
|
|
print(f"\n{Colors.BLUE}{Colors.BOLD}▶ {title}{Colors.ENDC}")
|
|
print(f"{Colors.DIM}{'─' * 50}{Colors.ENDC}")
|
|
|
|
|
|
def print_alert_info(alert: dict):
|
|
"""Print alert payload info"""
|
|
print(f" {Colors.YELLOW}告警類型:{Colors.ENDC} {alert['alert_type']}")
|
|
print(f" {Colors.YELLOW}嚴重度:{Colors.ENDC} {alert['severity']}")
|
|
print(f" {Colors.YELLOW}目標資源:{Colors.ENDC} {alert['target_resource']}")
|
|
print(f" {Colors.YELLOW}命名空間:{Colors.ENDC} {alert['namespace']}")
|
|
print(f" {Colors.YELLOW}訊息:{Colors.ENDC} {alert['message']}")
|
|
if alert.get('metrics'):
|
|
print(f" {Colors.YELLOW}指標:{Colors.ENDC}")
|
|
for k, v in alert['metrics'].items():
|
|
print(f" • {k}: {v}")
|
|
|
|
|
|
def print_response(response: dict, status_code: int):
|
|
"""Print API response"""
|
|
if status_code == 200 and response.get('success'):
|
|
print(f"\n{Colors.GREEN}{Colors.BOLD}✓ 告警發射成功！{Colors.ENDC}")
|
|
print(f" {Colors.CYAN}Approval ID:{Colors.ENDC} {response.get('approval_id', 'N/A')}")
|
|
print(f" {Colors.CYAN}風險等級:{Colors.ENDC} {response.get('risk_level', 'N/A')}")
|
|
print(f" {Colors.CYAN}建議動作:{Colors.ENDC} {response.get('suggested_action', 'N/A')}")
|
|
print(f" {Colors.CYAN}聚合次數:{Colors.ENDC} {response.get('hit_count', 1)}")
|
|
if response.get('converged'):
|
|
print(f" {Colors.YELLOW}⚡ 告警已收斂 (跳過 LLM){Colors.ENDC}")
|
|
else:
|
|
print(f"\n{Colors.RED}{Colors.BOLD}✗ 告警發射失敗！{Colors.ENDC}")
|
|
print(f" {Colors.RED}狀態碼:{Colors.ENDC} {status_code}")
|
|
print(f" {Colors.RED}回應:{Colors.ENDC} {response}")
|
|
|
|
|
|
def print_footer():
|
|
"""Print footer with instructions"""
|
|
print(f"\n{Colors.DIM}{'─' * 50}{Colors.ENDC}")
|
|
print(f"{Colors.GREEN}📺 請查看戰情室前端：{Colors.ENDC} http://localhost:3000")
|
|
print(f"{Colors.GREEN}📋 右側面板應顯示新的 ApprovalCard{Colors.ENDC}")
|
|
print(f"{Colors.DIM}時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}{Colors.ENDC}\n")
|
|
|
|
|
|
# =============================================================================
|
|
# Main Logic
|
|
# =============================================================================
|
|
|
|
async def fire_alert(alert_type: str, severity: str | None = None) -> bool:
|
|
"""
|
|
發射測試告警
|
|
|
|
Args:
|
|
alert_type: 告警類型 (db_connection_timeout, k8s_pod_crash, etc.)
|
|
severity: 覆蓋嚴重度 (optional)
|
|
|
|
Returns:
|
|
bool: 是否成功
|
|
"""
|
|
# 取得告警場景
|
|
if alert_type not in ALERT_SCENARIOS:
|
|
print(f"{Colors.RED}❌ 未知告警類型: {alert_type}{Colors.ENDC}")
|
|
print(f"{Colors.DIM}可用類型: {', '.join(ALERT_SCENARIOS.keys())}{Colors.ENDC}")
|
|
return False
|
|
|
|
alert = ALERT_SCENARIOS[alert_type].copy()
|
|
|
|
# 覆蓋嚴重度
|
|
if severity:
|
|
alert['severity'] = severity
|
|
|
|
print_section("告警 Payload")
|
|
print_alert_info(alert)
|
|
|
|
print_section("發射告警至 Webhook API")
|
|
print(f" {Colors.CYAN}端點:{Colors.ENDC} {WEBHOOK_ENDPOINT}")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(
|
|
WEBHOOK_ENDPOINT,
|
|
json=alert,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
result = response.json()
|
|
print_response(result, response.status_code)
|
|
|
|
return response.status_code == 200
|
|
|
|
except httpx.ConnectError:
|
|
print(f"\n{Colors.RED}{Colors.BOLD}✗ 連線失敗！{Colors.ENDC}")
|
|
print(f" {Colors.RED}請確認後端 API 正在運行:{Colors.ENDC}")
|
|
print(f" {Colors.DIM}cd apps/api && uvicorn src.main:app --reload{Colors.ENDC}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"\n{Colors.RED}{Colors.BOLD}✗ 發生錯誤：{e}{Colors.ENDC}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""CLI Entry Point"""
|
|
parser = argparse.ArgumentParser(
|
|
description="🚀 AWOOOI Phase 2 導彈腳本 - 發射測試告警",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
可用告警類型:
|
|
db_connection_timeout PostgreSQL Database OOM (CRITICAL)
|
|
k8s_pod_crash Pod CrashLoopBackOff (MEDIUM)
|
|
high_cpu CPU Spike / Latency (MEDIUM)
|
|
disk_full Disk Full Warning (CRITICAL)
|
|
ssl_expiry SSL Certificate Expiry (LOW)
|
|
|
|
範例:
|
|
python -m scripts.fire_test_alert
|
|
python -m scripts.fire_test_alert --type db_connection_timeout
|
|
python -m scripts.fire_test_alert --type k8s_pod_crash --severity critical
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--type", "-t",
|
|
type=str,
|
|
default="db_connection_timeout",
|
|
choices=list(ALERT_SCENARIOS.keys()),
|
|
help="告警類型 (預設: db_connection_timeout)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--severity", "-s",
|
|
type=str,
|
|
choices=["info", "warning", "critical"],
|
|
help="覆蓋嚴重度 (預設使用場景預設值)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
print_banner()
|
|
success = asyncio.run(fire_alert(args.type, args.severity))
|
|
print_footer()
|
|
|
|
sys.exit(0 if success else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|