发布时间:2025-11-05 12:18:13 来源:创站工坊 作者:应用开发

希望在告警通知里有以下数据:
告知当前系统还有多少未处理的夜莺义告告警。告知当前告警恢复时候的自定具体值。告警通知里增加查看未处理告警的警模页面链接。要实现上面的夜莺义告需求很简单,夜莺监控的自定数据库表alert_cur_event保存了我们所需要的当前未处理的告警总数,而且夜莺监控也提供了查询未处理告警的警模面板,而对于告警恢复时候的夜莺义告值我们只需要根据自定义的恢复promql即可查询。
最简单的自定方式就是直接通过notify.py脚本进行告警发送,我们只需要做一丢丢修改即可。警模
整体脚本如下:
复制#!/usr/bin/env python # -*- coding: UTF-8 -*- import sys import json import requests import pymysql # 处理字符问题 reload(sys) sys.setdefaultencoding(utf-8) # 通过api查询指标 def getPrometheus(url,夜莺义告 promql): response = requests.get(url, params={query: promql}) data = json.loads(response.text) # 提取指标数据 if response.status_code == 200 result = data[data][result] if len(result) == 1: return result[0][value][1] else: return 0 else: return 0 def count_rows_and_get_rule_names(): try: conn = pymysql.connect( host=127.0.0.1, port=3306, user=n9e, passwd=1234, db=n9e_v6, charset=utf8mb4 ) cursor = conn.cursor() # Count the total number of rows in the table count_query = "SELECT COUNT(*) FROM alert_cur_event" cursor.execute(count_query) total_rows = cursor.fetchone()[0] return total_rows except Exception as e: print("Error: ", e) class Sender(object): @classmethod def send_qywx(cls, payload): users = payload.get(event).get("notify_users_obj") is_recovered = payload.get(event).get("is_recovered") tokens = {} phones = {} res = {} history_row = count_rows_and_get_rule_names() if is_recovered: # 获取自定义的恢复promql promQL = payload.get(event).get("annotations").get("recovery_promql") url = "http://127.0.0.1:9090/api/v1/query" res = getPrometheus(url, promQL) # 查询活跃告警的面板 currAlert = "http://127.0.0.1:17000/alert-cur-events" for u in users: if u.get("phone"): phones[u.get("phone")] = 1 contacts = u.get("contacts") if contacts.get("qywx_robot_token", ""): tokens[contacts.get("qywx_robot_token", "")] = 1 headers = { "Content-Type": "application/json;charset=utf-8", "Host": "qyapi.weixin.qq.com" } for t in tokens: url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={}".format(t) content = payload.get(tpls).get("qywx", "qywx not found") content = "# **当前环境的全部告警数**: %s" % (history_row) + "\n" + content if is_recovered: content = content + "\n" + "> **恢复时值**: %s" % (res) if history_row > 0: content = content + "\n" + "[当前活跃告警](%s)" % (currAlert) body = { "msgtype": "markdown", "markdown": { "content": content } } response = requests.post(url, headers=headers, data=json.dumps(body)) def main(): payload = json.load(sys.stdin) with open(".payload", w) as f: f.write(json.dumps(payload, indent=4)) for ch in payload.get(event).get(notify_channels): send_func_name = "send_{}".format(ch.strip()) if not hasattr(Sender, send_func_name): print("function: {} not found", send_func_name) continue send_func = getattr(Sender, send_func_name) send_func(payload) def hello(): print("hello nightingale") if __name__ == "__main__": if len(sys.argv) == 1: main() elif sys.argv[1] == "hello": hello() else: print("I am confused")1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.42.43.44.45.46.47.48.49.50.51.52.53.54.55.56.57.58.59.60.61.62.63.64.65.66.67.68.69.70.71.72.73.74.75.76.77.78.79.80.81.82.83.84.85.86.87.88.89.90.91.92.93.94.95.96.97.98.99.100.101.102.103.104.105.106.107.108.109.110.111.112.113.114.115.116.117.118.119.120.121.122.需要在服务器上安装pymysql以及requests包
然后将上面的脚本放到夜莺监控面板->系统设置->通知设置->通知脚本中,并将脚本设置为启用状态。自定

然后新增名叫qywx的警模通知媒介以及名叫qywx_robot_token的联系方式,在发送告警的夜莺义告时候会通过通知媒介来调用通知方法,比如你的自定通知媒介名叫zhangsan,服务器托管那么你定义的警模方法名就是send_zhangsan。另外还会从联系方式处获取发送的token。
然后我们来创建一个通知模板,这个模板是在原生的基础上进行更改的,如下创建一个名叫qywx的模板。
复制> **级别状态**: {{if .IsRecovered}}<font color="info">告警恢复</font>{{else}}<font color="warning">发生告警</font>{{end}} > **级别级别**: S{{.Severity}} > **告警类型**: {{.RuleName}}{{if .RuleNote}} > **告警详情**: {{.RuleNote}}{{end}}{{if .TargetIdent}} > **监控对象**: {{.TargetIdent}}{{end}} > **监控指标**: {{.TagsJSON}}{{if not .IsRecovered}} > **触发时值**: {{.TriggerValue}}{{end}} {{if .IsRecovered}}> **恢复时间**: {{timeformat .LastEvalTime}}{{else}}> **首次触发时间**: {{timeformat .FirstTriggerTime}}{{end}} {{$time_duration := sub now.Unix .FirstTriggerTime}}{{if .IsRecovered}}{{$time_duration = sub .LastEvalTime .FirstTriggerTime}}{{end}}> **距离首次告警**: {{humanizeDurationInterface $time_duration}} > **发送时间**: {{timestamp}}1.2.3.4.5.6.7.8.9.10.在实际发送过程中会对模板进行相应的增加。
最后,再来配置告警,比如我们现在要配置一个K8s中Pod的状态异常的告警规则,如下:

填写具体的规则名以及备注,并且填写具体的promql。
往下继续填写通知媒介以及附加信息。

其中附加信息中就有告警恢复时候的promql,在python脚本中会获取当前的promql,然后调用prometheus的接口进行查询当前值,最后填充到告警模板中去。
以上就是具体的b2b信息网实现思路,希望对你有所启发。
除了这种python脚本的方式,还可以通过自定义webhook的方式实现,夜莺是支持回调地址的,只需要把回调地址填写进去即可。
那这个webhook应该怎么开发呢?
其实不需要我们做啥大的开发,直接把夜莺的源码里告警相关的CV出来,改吧改吧就能直接用了。
首先,把alert_cur_event的数据结构弄过来,查表就查它。
其次,增加一个查询prometheus的接口,如下:
复制package prometheus import ( "context" "devops-webhook-service/src/server/config" "github.com/prometheus/client_golang/api" "github.com/prometheus/client_golang/api/prometheus/v1" "github.com/prometheus/common/model" "github.com/toolkits/pkg/logger" "time") func GetMetricsValue(promql string) string { client, err := api.NewClient(api.Config{ Address: config.C.Prometheus.Address, }) if err != nil { logger.Error("init prometheus client failed. err: ", err) } queryAPI := v1.NewAPI(client) result, warnings, err := queryAPI.Query(context.TODO(), promql, time.Now()) if err != nil { // handle error logger.Error("query prometheus metrics failed. err: ", err) } if len(warnings) > 0 { // handle warnings } vector := result.(model.Vector) //for _, sample := range vector { // fmt.Printf("Time: %v, Value: %v\n", sample.Timestamp, sample.Value) //} return vector[0].Value.String() }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.再则,我们就可以把需要的告警字段都动议到告警模板中,通过template自动填充数据了。
复制## 当前环境的告警总数: {{ .TotalAlert}} --- **级别状态**: {{if .IsRecovered}}<font color="info">S{{.Severity}} Recovered</font>{{else}}<font color="warning">S{{.Severity}} Triggered</font>{{end}} **规则标题**: {{.RuleName}}{{if .TargetIdent}} **监控对象**: {{.TargetIdent}}{{end}}{{ if .IsRecovered}} **当前值**: {{ .RecoveryValue}}{{end}} **监控指标**: {{.TagsJSON}}{{if not .IsRecovered}} **触发时值**: {{.TriggerValue}}{{end}} {{if .IsRecovered}}**恢复时间**: {{timeformat .LastEvalTime}}{{else}}**首次触发时间**: {{timeformat .TriggerTime}}{{end}} **发送时间**: {{timestamp}} ---1.2.3.4.5.6.7.8.9.10.11.最后,就是在notify.go中做一丢丢的小修改。
比如event事件增加两个字段。免费源码下载
复制type NoticeEvent struct { *models.AlertCurEvent RecoveryValue string // 恢复时候的值 TotalAlert int // 告警总数 }1.2.3.4.5.比如在notify.go中的GenNotice方法里,增加查询prometheus和数据库的代码。
复制var recoveryValue string if event.IsRecovered { text := event.RuleNote promql := strings.Split(text, "=")[1] recoveryValue = prometheus.GetMetricsValue(promql) } // 获取当前剩余的总告警数 events, err := models.AlertCurEventGetAll(event.Cluster) if err != nil { logger.Error("get alert event failed. err: ", err) }1.2.3.4.5.6.7.8.9.10.11.12.整体代码也就只需要一丢丢东西。

以上就是整体的实现了,这只是领导根据领导的需要做的,每个团队的需求不一样,实现方式肯定也不通,这里只是抛砖引玉。
个人建议使用webhook比较好一点,因为可以比较灵活的增加其他的功能,比如告警认领,比如告警抑制,比如告警转发等。
另外,最近刚换工作没多久,写的文章少了,但是对技术的热爱并没有减少。