notify.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import sys
import json
import os
import smtplib
import time
import requests
from email.mime.text import MIMEText
from email.header import Header
from bottle import template

reload(sys)                      # reload 才能调用 setdefaultencoding 方法
sys.setdefaultencoding('utf-8')  # 设置 'utf-8'

# 希望的demo实现效果：
# 1. 从stdin拿到告警信息之后，格式化为一个有缩进的json写入一个临时文件
# 2. 文件路径和名字是.alerts/${timestamp}_${ruleid}
# 3. 调用SMTP服务器发送告警，微信、钉钉、飞书、slack、jira、短信、电话等等留给社区实现

# 脚本二开指南
# 1. 可以根据下面的 TEST_ALERT_JSON 中的结构修改脚本发送逻辑，定制化告警格式见tpl下的模块文件
# 2. 每个告警会以json文件的格式存储在 LOCAL_EVENT_FILE_DIR 下面，文件名为 filename = '%d_%d_%d' % (rule_id, event_id, trigger_time)
# 3. 告警通道需要自行定义Send类中的send_xxx的方法，反射调用：举例 event.notify_channels = [邮件] 则需要在NOTIFY_CHANNEL_DICT配置 邮件=email，在Send类中实现 send_email 方法
# 4. im发群信息，比如钉钉发群信息需要群的webhook机器人 token，这个信息可以在user的contacts map中，各个send_方法处理即可
# 5. 用户创建一个虚拟的用户保存上述im群 的机器人token信息 user的contacts map中

mail_host = "smtp.163.com"
mail_port = 994
mail_user = "ulricqin"
mail_pass = "password"
mail_from = "ulricqin@163.com"

# 本地告警event json存储目录
LOCAL_EVENT_FILE_DIR = ".alerts"
NOTIFY_CHANNELS_SPLIT_STR = " "

# dingding 群机器人token 配置字段
DINGTALK_ROBOT_TOKEN_NAME = "dingtalk_robot_token"
DINGTALK_API = "https://oapi.dingtalk.com/robot/send"

WECOM_ROBOT_TOKEN_NAME = "wecom_robot_token"
WECOM_API = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"

NOTIFY_CHANNEL_DICT = {
  "email":"email",
  "sms":"sms",
  "voice":"voice",
  "dingtalk":"dingtalk",
  "wecom":"wecom"
}

# stdin 告警json实例
TEST_ALERT_JSON = {
    "event": {
        "alert_duration": 10,
        "notify_channels": "dingtalk",
        "res_classpaths": "all",
        "id": 4,
        "notify_group_objs": None,
        "rule_note": "",
        "history_points": [
            {
                "metric": "go_goroutines",
                "points": [
                    {
                        "t": 1625213114,
                        "v": 33.0
                    }
                ],
                "tags": {
                    "instance": "localhost:9090",
                    "job": "prometheus"
                }
            }
        ],
        "priority": 1,
        "last_sent": True,
        "tag_map": {
            "instance": "localhost:9090",
            "job": "prometheus"
        },
        "hash_id": "ecb258d2ca03454ee390a352913c461b",
        "status": 0,
        "tags": "instance=localhost:9090 job=prometheus",
        "trigger_time": 1625213114,
        "res_ident": "ident1",
        "rule_name": "alert_test",
        "is_prome_pull": 1,
        "notify_users": "1",
        "notify_groups": "",
        "runbook_url": "",
        "values": "[vector={__name__=\"go_goroutines\", instance=\"localhost:9090\", job=\"prometheus\"}]: [value=33.000000]",
        "readable_expression": "go_goroutines>0",
        "notify_user_objs": None,
        "is_recovery": 1,
        "rule_id": 1
    },
    "rule": {
        "alert_duration": 10,
        "notify_channels": "dingtalk",
        "enable_stime": "00:00",
        "id": 1,
        "note": "",
        "create_by": "root",
        "append_tags": "",
        "priority": 1,
        "update_by": "root",
        "type": 1,
        "status": 0,
        "recovery_notify": 0,
        "enable_days_of_week": "1 2 3 4 5 6 7",
        "callbacks": "localhost:10000",
        "notify_users": "1",
        "notify_groups": "",
        "runbook_url": "",
        "name": "a",
        "update_at": 1625211576,
        "create_at": 1625211576,
        "enable_etime": "23:59",
        "group_id": 1,
        "expression": {
            "evaluation_interval": 4,
            "promql": "go_goroutines>0"
        }
    },
    "users": [
        {
            "username": "root",
            "status": 0,
            "contacts": {
                "dingtalk_robot_token": "xxxxxx"
            },
            "create_by": "system",
            "update_at": 1625211432,
            "create_at": 1624871926,
            "email": "",
            "phone": "",
            "role": "Admin",
            "update_by": "root",
            "portrait": "",
            "nickname": "\u8d85\u7ba1",
            "id": 1
        }
    ]
}


def main():
    payload = json.load(sys.stdin)
    trigger_time = payload['event']['trigger_time']
    event_id = payload['event']['id']
    rule_id = payload['rule']['id']
    notify_channels = payload['event'].get('notify_channels').strip().split(NOTIFY_CHANNELS_SPLIT_STR)
    if len(notify_channels) == 0:
        msg = "notify_channels_empty"
        print(msg)
        return
    # 持久化到本地json文件
    persist(payload, rule_id, event_id, trigger_time)
    # 生成告警内容
    alert_content = sms_content_gen(values_gen(payload))
    for ch in notify_channels:
        send_func_name = "send_{}".format(NOTIFY_CHANNEL_DICT.get(ch.strip()))
        has_func = hasattr(Send, send_func_name)

        if not has_func:
            msg = "[send_func_name_err][func_not_found_in_Send_class:{}]".format(send_func_name)
            print(msg)
            continue
        send_func = getattr(Send, send_func_name)
        send_func(alert_content, payload)

def values_gen(payload):
    event_obj = payload.get("event")
    values = {
        "IsAlert": event_obj.get("is_recovery") == 0,
        "IsMachineDep": event_obj.get("res_classpaths") != "",
        "Status": status_gen(event_obj.get("priority"),event_obj.get("is_recovery")),
        "Sname": event_obj.get("rule_name"),
        "Ident": event_obj.get("res_ident"),
        "Classpath": event_obj.get("res_classpaths"),
        "Metric": metric_gen(event_obj.get("history_points")),
        "Tags": event_obj.get("tags"),
        "Value": event_obj.get("values"),
        "ReadableExpression": event_obj.get("readable_expression"),
        "TriggerTime": time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(event_obj.get("trigger_time"))),
        "Elink": "http://n9e.didiyun.com/strategy/edit/{}".format(event_obj.get("rule_id")),
        "Slink": "http://n9e.didiyun.com/event/{}".format(event_obj.get("id"))
    }

    return values

def email_content_gen(values):
    return template('etc/script/tpl/mail.tpl', values)

def sms_content_gen(values):
    return template('etc/script/tpl/sms.tpl', values)

def status_gen(priority,is_recovery):
    is_recovery_str_m = {1: "恢复", 0: "告警"}
    status = "P{} {}".format(priority, is_recovery_str_m.get(is_recovery))
    return status

def subject_gen(priority,is_recovery,rule_name):
    is_recovery_str_m = {1: "恢复", 0: "告警"}
    subject = "P{} {} {}".format(priority, is_recovery_str_m.get(is_recovery), rule_name)
    return subject

def metric_gen(history_points):
    metrics = [] 
    for item in history_points:
        metrics.append(item.get("metric"))
    return ",".join(metrics)

def persist(payload, rule_id, event_id, trigger_time):
    if not os.path.exists(LOCAL_EVENT_FILE_DIR):
        os.makedirs(LOCAL_EVENT_FILE_DIR)

    filename = '%d_%d_%d' % (rule_id, event_id, trigger_time)
    filepath = os.path.join(LOCAL_EVENT_FILE_DIR, filename)
    with open(filepath, 'w') as f:
        f.write(json.dumps(payload, indent=4))


class Send(object):
    @classmethod
    def send_email(cls, alert_content, payload):
        users = payload.get("users")
        emails = [x.get("email") for x in users]
        if not emails:
            return

        recipients = emails
        mail_body = email_content_gen(values_gen(payload))
        message = MIMEText(mail_body, 'html', 'utf-8')
        message['From'] = mail_from
        message['To'] = ", ".join(recipients)
        message["Subject"] = subject_gen(payload.get("event").get("priority"),payload.get("event").get("is_recovery"),payload.get("event").get("rule_name"))

        smtp = smtplib.SMTP_SSL(mail_host, mail_port)
        smtp.login(mail_user, mail_pass)
        smtp.sendmail(mail_from, recipients, message.as_string())
        smtp.close()

        print("send_mail_success")

    @classmethod
    def send_wecom(cls, alert_content, payload):
        users = payload.get("users")

        for u in users:
            contacts = u.get("contacts")
            wecom_robot_token = contacts.get(WECOM_ROBOT_TOKEN_NAME, "")

            if wecom_robot_token == "":
                continue

            wecom_api_url = "{}?key={}".format(WECOM_API, wecom_robot_token)
            atMobiles = [u.get("phone")]
            headers = {'Content-Type': 'application/json;charset=utf-8'}
            payload = {
                "msgtype": "text",
                "text": {
                    "content": alert_content
                },
                "at": {
                    "atMobiles": atMobiles,
                    "isAtAll": False
                }
            }
            res = requests.post(wecom_api_url, json.dumps(payload), headers=headers)
            print(res.status_code)
            print(res.text)
            print("send_wecom")


    @classmethod
    def send_dingtalk(cls, alert_content, payload):
        # 钉钉发群信息需要群的webhook机器人 token，这个信息可以在user的contacts map中

        users = payload.get("users")

        for u in users:
            contacts = u.get("contacts")

            dingtalk_robot_token = contacts.get(DINGTALK_ROBOT_TOKEN_NAME, "")

            if dingtalk_robot_token == "":
                print("dingtalk_robot_token_not_found")
                continue

            dingtalk_api_url = "{}?access_token={}".format(DINGTALK_API, dingtalk_robot_token)
            atMobiles = [u.get("phone")]
            headers = {'Content-Type': 'application/json;charset=utf-8'}
            payload = {
                "msgtype": "text",
                "text": {
                    "content": alert_content
                },
                "at": {
                    "atMobiles": atMobiles,
                    "isAtAll": False
                }
            }
            res = requests.post(dingtalk_api_url, json.dumps(payload), headers=headers)
            print(res.status_code)
            print(res.text)

            print("send_dingtalk")


def mail_test():
    print("mail_test_todo")

    recipients = ["ulricqin@qq.com", "ulric@163.com"]

    payload =  json.loads(json.dumps(TEST_ALERT_JSON))
    mail_body = email_content_gen(values_gen(payload))
    message = MIMEText(mail_body, 'html', 'utf-8')
    message['From'] = mail_from
    message['To'] = ", ".join(recipients)
    message["Subject"] = subject_gen(payload.get("event").get("priority"),payload.get("event").get("is_recovery"),payload.get("event").get("rule_name"))

    smtp = smtplib.SMTP_SSL(mail_host, mail_port)
    smtp.login(mail_user, mail_pass)
    smtp.sendmail(mail_from, recipients, message.as_string())
    smtp.close()

    print("mail_test_done")

if __name__ == "__main__":
    if len(sys.argv) == 1:
        main()
    elif sys.argv[1] == "mail":
        mail_test()
    else:
        print("I am confused")