前言
前面写过使用夜莺作为Prometheus告警引擎的文章，在此基础上接入DeepSeek进行故障分析，给出可能引起故障的原因和排查方法，大大降低平均故障处理时间（MTTR）。
工作流程：Prometheus周期采集Exporter的指标并存储在本地，夜莺周期查询Prometheus中的指标是否符合故障告警规则，在产生故障后将故障信息发送给DeepSeek，DeepSeek通过分析给出的故障原因和处理建议，拼接原告警信息和DeepSeek分析结果一并发给用户。对于没有使用夜莺的小伙伴可以直接使用webhook来执行脚本。
「流程图」
1. 新建通知模板
告警通知-->通知模板-->新增, 新增aiops模板
#### {{if .IsRecovered}}<font color="#008800">? {{.RuleName}}恢复</font>{{else}}<font color="#FF0000">? {{.RuleName}}告警</font>{{end}}
---
**级别状态**：{{if .IsRecovered}}<fontcolor="#008800">S{{.Severity}}</font>{{else}}<fontcolor="#FF0000">S{{.Severity}}</font>{{end}}
{{if eq (index .TagsMap "job") "web_status"}}
**归属公司**：{{index .TagsMap "company"}}
**项目名称**：{{index .TagsMap "project_cn"}}
**系统名称**：{{index .TagsMap "name"}}
{{if .IsRecovered}}**恢复内容**：{{index .TagsMap "name"}} 当前已恢复正常！
{{else}}**告警内容**：{{index .TagsMap "name"}} 当前无法访问！
{{end}}
**系统地址**：[{{index .TagsMap "instance"}}]({{index .TagsMap "instance"}})
{{end}}
{{if .IsRecovered}}**触发时间**：{{timeformat .FirstTriggerTime}}
**恢复时间**：{{timeformat .LastEvalTime}}{{else}}**触发时间**：{{timeformat .FirstTriggerTime}}{{end}}
2. 新建通知媒介
告警通知-->通知设置-->通知媒介-->添加
名称： aiops标识： aiops
3. 配置通知脚本
告警通知-->通知设置-->通知脚本-->使用脚本
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
importsys
importjson
importrequests
classSender(object):
@classmethod
defsend_email(cls, payload):
# already done in go code
pass
@classmethod
defsend_wecom(cls, payload):
# already done in go code
pass
# 钉钉机器人
DINGTALK_URL ="https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXX"
# DeepSeek key
DEEPSEEK_URL ="https://api.deepseek.com/v1/chat/completions"
DEEPSEEK_KEY ="sk-XXXXXXXXXX"
@classmethod
defcall_deepseek(cls, message):
headers = {
"Content-Type":"application/json",
"Authorization":f"Bearer{cls.DEEPSEEK_KEY}"
}
data = {
"model":"deepseek-chat",
"messages": [{
"role":"user",
"content":f"""
告警信息：{message}
你是一名运维领域的专家，请分析告警信息给出可能原因、处理建议和紧急程度
排版要求：AI故障分析标题为蓝色h4大小，语言简洁突出重点
"""
}]
}
try:
response = requests.post(cls.DEEPSEEK_URL, headers=headers, json=data)
response.raise_for_status()
returnresponse.json()['choices'][0]['message']['content']
exceptExceptionase:
print(f"Deepseek API error:{str(e)}")
return"无法获取处理建议"
@classmethod
defsend_dingtalk(cls, payload):
original_message = payload.get('tpls').get("dingtalk","dingtalk not found")
analysis = cls.call_deepseek(original_message)
final_message =f"""{original_message}
---
{analysis}
"""
headers = {
"Content-Type":"application/json;charset=utf-8"
}
body = {
"msgtype":"markdown",
"markdown": {
"title":"告警通知",
"text": final_message
}
}
response = requests.post(cls.DINGTALK_URL, headers=headers, data=json.dumps(body))
print(f"notify_dingtalk: status_code={response.status_code}response_text={response.text}")
@classmethod
defsend_mm(cls, payload):
# already done in go code
pass
@classmethod
defsend_sms(cls, payload):
pass
@classmethod
defsend_voice(cls, payload):
pass
defmain():
payload = json.load(sys.stdin)
withopen(".payload",'w')asf:
f.write(json.dumps(payload, indent=4))
forchinpayload.get('event').get('notify_channels'):
send_func_name ="send_{}".format(ch.strip())
ifnothasattr(Sender, send_func_name):
print("function: {} not found", send_func_name)
continue
send_func = getattr(Sender, send_func_name)
send_func(payload)
defhello():
print("hello nightingale")
if__name__ =="__main__":
iflen(sys.argv) ==1:
main()
elifsys.argv[1] =="hello":
hello()
else:
print("I am confused")
DeepSeek 分析回答需要一定时间，超时时间建议配置 120s。
4. 告警规则启用 AIOps 通知方式
告警管理-->告警规则，原有的规则勾上aiops通知媒介
5. 告警测试
原告警信息
接入大模型后的告警信息
在这里只是简单的将告警信息发送给DeepSeek大模型进行分析然后将分析的结果发送给我们，分析的结果一般没办法完全符合我们的预期。下一步需要进行「RAG」(检索增强生成)，通过检索外部知识库的方式，将知识库中的相关内容作为提示输入给大模型，从而给出更加符合我们预期的分析结果。在AIOps建设的过程中，运维知识库的建设将非常重要，「RAG」的结果符合我们的预期后，再结合工作流、AI Agent的方式进行告警收敛、故障自愈等操作。相关开源工具 ---dify