attacker
5333f6f04a
- 后端: NotifyChannel 模型新增 feishu 类型, FeishuWebhookConfig - 后端: TestNotifyChannel 真实发送飞书测试消息(文本格式) - 后端: 告警引擎触发/恢复时自动发送飞书通知(sendFeishuAlert/sendFeishuRecovery) - 后端: AlertEngine 新增 tcp_time_wait/tcp_close_wait/service_process 指标检测 - 后端: NetworkStatModel 新增 TCPCloseWait 字段 - 后端: QuickSetupRules 一键创建6条告警规则(CPU/内存/磁盘/TIME_WAIT/CLOSE_WAIT/进程) - 后端: 注册 /alerts/quick-setup 路由 - 前端: AlertChannelDialog 新增飞书类型选项 - 前端: helpers 新增 getChannelIcon/getChannelTypeLabel 支持 feishu/dingtalk/wechat
123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
设置飞书Webhook通知渠道 + 创建告警规则 + 测试
|
|
"""
|
|
import urllib.request
|
|
import json
|
|
import sys
|
|
import time
|
|
|
|
BASE = "http://10.10.10.75"
|
|
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/4c9b0763-fb70-448b-8ece-2d2fbcf5e3ee"
|
|
|
|
def api(method, path, data=None):
|
|
url = f"{BASE}{path}"
|
|
body = json.dumps(data).encode() if data else None
|
|
req = urllib.request.Request(url, data=body, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
if jwt_token:
|
|
req.add_header("Authorization", f"Bearer {jwt_token}")
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=15)
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
print(f" HTTP {e.code}: {e.read().decode()[:200]}")
|
|
return None
|
|
|
|
# Step 1: Login
|
|
print("=" * 60)
|
|
print("Step 1: 登录获取 JWT Token")
|
|
print("=" * 60)
|
|
resp = api("POST", "/api/v1/auth/login", {"username": "admin", "password": "123456"})
|
|
if not resp or resp.get("code") != 200:
|
|
print("ERROR: 登录失败")
|
|
sys.exit(1)
|
|
jwt_token = resp.get("data", {}).get("token", "")
|
|
if not jwt_token:
|
|
# Try different key
|
|
jwt_token = resp.get("token", "")
|
|
print(f" Token: {jwt_token[:20]}...")
|
|
print()
|
|
|
|
# Step 2: Create Feishu Webhook Notify Channel
|
|
print("=" * 60)
|
|
print("Step 2: 创建飞书 Webhook 通知渠道")
|
|
print("=" * 60)
|
|
channel_config = json.dumps({"url": FEISHU_WEBHOOK})
|
|
resp = api("POST", "/alerts/notify-channels", {
|
|
"name": "飞书运维告警群",
|
|
"type": "feishu",
|
|
"config": channel_config
|
|
})
|
|
if resp and resp.get("code") == 200:
|
|
channel_id = resp["data"]["id"]
|
|
print(f" 渠道创建成功: ID={channel_id}")
|
|
else:
|
|
print(f" 创建结果: {resp}")
|
|
|
|
# Also list existing channels
|
|
resp = api("GET", "/alerts/notify-channels")
|
|
if resp:
|
|
channels = resp.get("data", {}).get("list", [])
|
|
print(f" 现有渠道: {len(channels)} 个")
|
|
for ch in channels:
|
|
print(f" [{ch['type']}] {ch['name']} (enabled={ch['enabled']}) id={ch['id']}")
|
|
if channels:
|
|
channel_id = channels[0]["id"]
|
|
print()
|
|
|
|
# Step 3: Test the Feishu channel
|
|
print("=" * 60)
|
|
print("Step 3: 测试飞书通知渠道")
|
|
print("=" * 60)
|
|
if channel_id:
|
|
resp = api("POST", f"/alerts/notify-channels/{channel_id}/test")
|
|
if resp:
|
|
print(f" 测试结果: {resp.get('message', resp)}")
|
|
print(f" 状态: {resp.get('data', {}).get('status', 'unknown')}")
|
|
else:
|
|
print(" 跳过: 没有渠道ID")
|
|
print()
|
|
|
|
# Step 4: Create alert rules
|
|
print("=" * 60)
|
|
print("Step 4: 创建告警规则 (一键报警)")
|
|
print("=" * 60)
|
|
|
|
# Use the quick-setup endpoint
|
|
resp = api("POST", "/alerts/quick-setup", {
|
|
"rules": [
|
|
{"name": "CPU使用率", "metric": "cpu", "operator": ">", "threshold": 80},
|
|
{"name": "内存使用率", "metric": "mem", "operator": ">", "threshold": 80},
|
|
{"name": "磁盘使用率", "metric": "disk", "operator": ">", "threshold": 90},
|
|
{"name": "TIME_WAIT连接数", "metric": "tcp_time_wait", "operator": ">", "threshold": 15000},
|
|
{"name": "CLOSE_WAIT连接数", "metric": "tcp_close_wait", "operator": ">", "threshold": 200},
|
|
{"name": "进程异常退出", "metric": "service_process", "operator": "==", "threshold": 0},
|
|
]
|
|
})
|
|
if resp:
|
|
print(f" 结果: {resp.get('message', resp)}")
|
|
print(f" 详情: {resp.get('data', {})}")
|
|
|
|
# List all rules
|
|
resp = api("GET", "/alerts/rules")
|
|
if resp:
|
|
rules = resp.get("data", {}).get("list", [])
|
|
print(f"\n 现有告警规则: {len(rules)} 条")
|
|
for r in rules:
|
|
print(f" [{r.get('type', '-')}] {r.get('name', '-')} (enabled={r.get('enabled', '-')})")
|
|
|
|
# Also check agent rules
|
|
resp = api("GET", "/api/v1/agent/rules")
|
|
if resp and resp.get("code") == 200:
|
|
rules = resp.get("data", [])
|
|
print(f"\n Agent 告警规则: {len(rules) if isinstance(rules, list) else 0} 条")
|
|
if isinstance(rules, list):
|
|
for r in rules[:10]:
|
|
print(f" [{r.get('metric', '-')}] {r.get('name', '-')} {'启用' if r.get('enabled') else '禁用'}")
|
|
|
|
print()
|
|
print("=" * 60)
|
|
print("完成！检查 http://10.10.10.75/alert-center 查看告警中心和规则")
|
|
print("=" * 60)
|