夜莺-Nightingale
夜莺V6
项目介绍 架构介绍
快速开始 快速开始
黄埔营
安装部署 安装部署
升级
采集器 采集器
使用手册 使用手册
API API
数据库表结构 数据库表结构
FAQ FAQ
开源生态
Prometheus
版权声明
第1章:天降奇兵 第1章:天降奇兵
第2章:探索PromQL 第2章:探索PromQL
第3章:Prometheus告警处理 第3章:Prometheus告警处理
第4章:Exporter详解 第4章:Exporter详解
第5章:数据与可视化 第5章:数据与可视化
第6章:集群与高可用 第6章:集群与高可用
第7章:Prometheus服务发现 第7章:Prometheus服务发现
第8章:监控Kubernetes 第8章:监控Kubernetes
第9章:Prometheus Operator 第9章:Prometheus Operator
参考资料

聯繫方式

聯繫方式主要用途有兩個

  1. 可以控制用戶信息中展示哪些聯繫方式,把不會用到的聯繫方式可以關閉
  2. 有時候用戶自己新增加的通知媒介,比如公司內部的im,需要配置用戶的聯繫方式,可以在這裡創建

聯繫方式1

新增聯繫方式後可以在創建用戶-聯繫方式選擇。

聯繫方式2

使用場景

用戶自己創建的通知媒介和聯繫方式主要在自定義通知腳本中使用,下面是一個腳本調用舉例,你可以參考樣例腳本,對接自己的內部通信工具。

系統會把告警事件的內容encode成json,然後通過stdin的方式傳給通知腳本,腳本示例如下:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import sys
import json
import urllib2

reload(sys)
sys.setdefaultencoding('utf8')  # 設置默認編碼為utf-8

class Sender(object):
    @classmethod
    # 方法send_newtalk,newtalk要和配置的通知媒介一致;
    def send_newtalk(cls, payload):
        try:
            #print("開始執行send_newtalk".encode('utf-8'))
            # 從payload中獲取事件數據
            event = payload.get('event')
            # 獲取通知用戶對象
            users = event.get("notify_users_obj")
            # 獲取告警規則名稱
            rule_name = event.get("rule_name")
            # 默認事件狀態為"觸發"
            event_state = "已觸發"
            # 如果事件已恢復,則狀態為"恢復"
            if event.get("is_recovered"):
                event_state = "已恢復"
            # 用於存儲釘釘機器人令牌和用戶電話
            tokens = {}
            phones = {}

            #print("提取用戶信息".encode('utf-8'))
            # 遍歷用戶,收集電話和釘釘令牌
            for u in users:
                if u.get("phone"):
                    phones[u.get("phone")] = 1

                contacts = u.get("contacts")
                # dingtalk_robot_token,需要根據需求替換對應名稱,在夜鶯通知設置-聯繫方式設定名稱,用戶裡面配置參數
                if contacts.get("dingtalk_robot_token", ""):
                    tokens[contacts.get("dingtalk_robot_token", "")] = 1
            # 設置請求頭
            headers = {
                "Content-Type": "application/json;charset=utf-8"
            }

            #print("令牌: {}".format(tokens).encode('utf-8'))
            #print("電話: {}".format(phones).encode('utf-8'))
            # 遍歷令牌,構建請求並發送到釘釘
            for t in tokens:
                url = "https://oapi.dingtalk.com/robot/send?access_token={}".format(t)
                body = {
                    "msgtype": "markdown",
                    "markdown": {
                        "title": "{} - {}".format(event_state, rule_name),
                        "text": "{} {}".format(payload.get('tpls').get("dingtalk.tpl", "找不到dingtalk.tpl"), ' '.join(["@"+i for i in phones.keys()]))
                    },
                    "at": {
                        "atMobiles": list(phones.keys()),
                        "isAtAll": False
                    }
                }

                #print("發送請求到URL: {}".format(url).encode('utf-8'))
                #print("請求主體: {}".format(body).encode('utf-8'))

                data = json.dumps(body)
                req = urllib2.Request(url, data=data, headers=headers)
                req.get_method = lambda: "POST"
                response = urllib2.urlopen(req)
                result = response.read()
                #print("回應狀態碼: {}".format(response.getcode()).encode('utf-8'))
                #print("回應文本: {}".format(result).encode('utf-8'))
        # 捕獲所有異常,防止程序崩潰
        except Exception as e:
            #print("send_newtalk中出錯: {}".format(e).encode('utf-8'))

def main():
    try:
        #print("從stdin讀取payload".encode('utf-8'))
        payload = json.load(sys.stdin)
        with open(".payload", 'w') as f:
            f.write(json.dumps(payload, indent=4))
        #print("payload已寫入.payload文件".encode('utf-8'))

        for ch in payload.get('event').get('notify_channels'):
            send_func_name = "send_{}".format(ch.strip())
            #print("處理通道: {}, 函數名稱: {}".format(ch, send_func_name).encode('utf-8'))
            if not hasattr(Sender, send_func_name):
                #print("找不到函數 {}".format(send_func_name).encode('utf-8'))
                continue
            send_func = getattr(Sender, send_func_name)
            send_func(payload)

    except Exception as e:
        print("main中出錯: {}".format(e).encode('utf-8'))

def hello():
    print("你好,夜鶯".encode('utf-8'))

if __name__ == "__main__":
    if len(sys.argv) == 1:
        main()
    elif sys.argv[1] == "hello":
        hello()
    else:
        print("我感到困惑".encode('utf-8'))

示例腳本中有很多 print 打印的輸出,主要是為了方便調試使用。如果開啟,可通過 n9e 的日誌進行查看,日誌調試在 INFO.log,報錯狀態在 ERROR.log,過濾關鍵詞 event_script_notify

腳本日誌

快猫星云 联系方式 快猫星云 联系方式
快猫星云 联系方式
快猫星云 联系方式
快猫星云 联系方式
快猫星云
OpenSource
开源版
Flashcat
Flashcat
FlashDuty
Flashduty