2026/7/23大约 4 分钟
核心流程设计
当告警发生时，不再仅仅是发邮件给 AI，而是先尝试执行自愈逻辑：
- 识别标签：Webhook 收到告警，判断
alertname或severity。 - 执行指令：匹配到
service_down时，调用系统命令运行ansible-playbook。 - 二次检查：重启后调用 Prometheus API 确认指标是否恢复。
- 最终通知：将“自愈结果”反馈到你的手机。
编写 Ansible 自愈剧本 (restart_service.yml)
路径/ansible/self_healing.yml
---
- name: ZH-Kinger 故障自愈任务
hosts: "{{ target_host }}"
become: yes
tasks:
- name: 强制重启目标服务
systemd:
name: "{{ target_service }}"
state: restarted
register: restart_res
- name: 验证服务状态
shell: "systemctl is-active {{ target_service }}"
register: service_check
- name: 输出自愈状态
debug:
msg: "服务 {{ target_service }} 现在的状态是: {{ service_check.stdout }}"升级中间件脚本
V3.0 核心升级简述
- 实现故障自愈 (Self-Healing) 不再停留在“发现问题”，而是直接通过 Ansible 介入。当 131/132 节点服务宕机时，系统会自动尝试重启修复，实现了运维的闭环自动化。
- AI 诊断具备“上下文意识” 你将“自愈结果”喂给了百炼 AI。AI 现在知道服务是否已经重启成功，并能根据结果给出差异化建议（成功则分析诱因，失败则给出人工抢修步骤）。
- 数据清洗与精准投放 新增了 IP 自动提取逻辑，能自动剔除 Prometheus 标签中的端口号（如
:9100），确保自愈指令能准确送达目标主机 IP。
import logging
from logging.handlers import RotatingFileHandler
import dashscope
from dashscope import Application
from flask import Flask, request
import smtplib
import time
import random
import subprocess # 新增：用于调用系统Ansible命令
from email.mime.text import MIMEText
from email.header import Header
# --- 1. 日志系统配置 ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AI-SelfHealing")
file_handler = RotatingFileHandler('/llm/ai_running.log', maxBytes=5*1024*1024, backupCount=3)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
app = Flask(__name__)
# --- 2. 核心参数配置 ---
dashscope.api_key = "你的API-KEY"
APP_ID = "你的应用id"
SMTP_SERVER = "smtp.163.com"
SMTP_PORT = 465
MAIL_USER = "www914132612@163.com"
MAIL_PASS = "你的邮箱授权码"
RECEIVER = "914132612@qq.com"
# --- 3. 自愈逻辑函数 (第四阶段核心) ---
def run_self_healing(instance_ip, alert_name):
"""匹配告警并执行Ansible自愈脚本"""
# 告警名与系统服务的映射表
service_map = {
"KubeletDown": "kubelet",
"DockerDown": "docker",
"NginxDown": "nginx",
"ServiceDown": "docker" # 默认策略
}
# 匹配服务名
target_svc = None
for key in service_map:
if key in alert_name:
target_svc = service_map[key]
break
if not target_svc:
return "⚠️ 未匹配到预设自愈方案，跳过自动修复。"
logger.info(f"🛠️ 触发自愈：尝试重启 {instance_ip} 上的 {target_svc} 服务...")
# 构造Ansible指令 (指向你指定的路径)
ansible_cmd = [
"ansible-playbook",
"/ansible/self_healing.yml",
"-e", f"target_host={instance_ip} target_service={target_svc}"
]
try:
# 执行命令，超时45秒
result = subprocess.run(ansible_cmd, capture_output=True, text=True, timeout=45)
if result.returncode == 0:
logger.info(f"✅ 节点 {instance_ip} 自愈指令执行成功")
return f"✅ 自愈动作已执行：已下发 {target_svc} 重启指令。"
else:
logger.error(f"❌ Ansible自愈失败: {result.stderr}")
return f"❌ 自愈动作失败：Ansible执行异常，请人工介入。"
except Exception as e:
logger.error(f"🚨 自愈模块崩溃: {str(e)}")
return f"🚨 自愈系统故障: {str(e)}"
def send_email(subject, content):
"""发送邮件并记录日志"""
message = MIMEText(content, 'plain', 'utf-8')
message['From'] = MAIL_USER
message['To'] = RECEIVER
message['Subject'] = Header(subject, 'utf-8')
try:
smtp_obj = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT)
smtp_obj.login(MAIL_USER, MAIL_PASS)
smtp_obj.sendmail(MAIL_USER, [RECEIVER], message.as_string())
smtp_obj.quit()
logger.info(f"邮件成功发送至 {RECEIVER}")
return True
except Exception as e:
logger.error(f"邮件发送异常: {str(e)}")
return False
@app.route('/webhook', methods=['POST'])
def webhook():
data = request.json
alerts = data.get('alerts', [])
logger.info(f"收到 Webhook 信号，包含 {len(alerts)} 条告警")
for alert in alerts:
# 提取干净的 IP (去掉 :9100 等端口号)
raw_instance = alert['labels'].get('instance', '192.168.31.x')
instance = raw_instance.split(':')[0]
alert_name = alert['labels'].get('alertname', '未知告警')
# --- A. 执行自愈流程 ---
healing_result = run_self_healing(instance, alert_name)
# --- B. 准备 AI 诊断 ---
current_session_id = f"session_{instance.replace('.', '_')}"
random_mark = random.randint(1000, 9999)
timestamp = time.strftime("%H:%M:%S")
logger.info(f"开始诊断节点 {instance} (Mark: {random_mark})...")
try:
# 将自愈结果直接喂给 AI 增加上下文
response = Application.call(
app_id=APP_ID,
prompt=(
f"【ZH-Kinger 自动化报告 - 编号：{random_mark}】\n"
f"1. 节点信息：{instance}\n"
f"2. 告警事件：{alert_name}\n"
f"3. 自动修复结果：{healing_result}\n\n"
f"请作为 SRE 专家：\n"
f"- 如果修复成功，分析该服务为何会崩溃（根因追溯）。\n"
f"- 如果修复失败，给出最急迫的人工介入指令。"
),
session_id=current_session_id,
parameters={'temperature': 0.8, 'top_p': 0.95}
)
if response.status_code == 200:
ai_report = response.output.text
# 4. 构造邮件内容 (包含自愈详情)
subject = f"【ZH-Kinger 自愈&诊断】{alert_name} @ {instance} (#{random_mark})"
content = (
f"告警节点: {instance}\n"
f"告警项目: {alert_name}\n"
f"自愈尝试: {healing_result}\n"
f"诊断序列: {random_mark}\n"
f"-------------------------------------------\n"
f"AI 专家深度分析:\n{ai_report}"
)
send_email(subject, content)
else:
logger.error(f"百炼 API 异常: {response.message}")
except Exception as e:
logger.error(f"处理告警逻辑时发生崩溃: {str(e)}")
return "OK", 200
if __name__ == '__main__':
logger.info("ZH-Kinger AI 闭环系统 V3.0 上线 (已集成Ansible自愈模块)")
app.run(host='0.0.0.0', port=5000)