夜莺-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
参考资料

Notification Channels

Notification channels have two main functions:

  1. Control which notification channels are displayed in alert rules, allowing you to disable unused contact methods for a more streamlined notification channel configuration.
  2. Users can create new notification channels to adapt to their company’s own notification systems. Custom notification scripts can then use these new channels to send notifications accordingly.

Notification Channels

Configuration Details

Name: Display name in the frontend

Identifier: Unique identifier for invocation

Hidden: Whether to display in the notification configuration of alert rules

Enabled: Whether this notification channel is enabled

Note: If not enabled, it will not be displayed in the alert rule configuration by default.

Use Cases

User-created notification channels and contact methods are primarily used in custom notification scripts. Below is an example of script invocation, which you can refer to when integrating with your internal communication tools.

The system will encode the content of the alert event as JSON and pass it to the notification script via stdin. Here’s a sample script:

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

reload(sys)
sys.setdefaultencoding('utf8')  # Set default encoding to utf-8

class Sender(object):
    @classmethod
    # The method send_newtalk, where newtalk should match the configured notification channel
    def send_newtalk(cls, payload):
        try:
            #print("Starting send_newtalk".encode('utf-8'))
            # Get event data from payload
            event = payload.get('event')
            # Get notified user objects
            users = event.get("notify_users_obj")
            # Get alert rule name
            rule_name = event.get("rule_name")
            # Default event state is "Triggered"
            event_state = "Triggered"
            # If the event has recovered, state is "Recovered"
            if event.get("is_recovered"):
                event_state = "Recovered"
            # Store Dingtalk robot tokens and user phones
            tokens = {}
            phones = {}

            #print("Extracting user information".encode('utf-8'))
            # Iterate through users, collecting phones and Dingtalk tokens
            for u in users:
                if u.get("phone"):
                    phones[u.get("phone")] = 1

                contacts = u.get("contacts")
                # dingtalk_robot_token, replace with the corresponding name as needed,
                # set the name in Nightingale notification settings - contact methods, configure parameters in user settings
                if contacts.get("dingtalk_robot_token", ""):
                    tokens[contacts.get("dingtalk_robot_token", "")] = 1
            # Set request headers
            headers = {
                "Content-Type": "application/json;charset=utf-8"
            }

            #print("Tokens: {}".format(tokens).encode('utf-8'))
            #print("Phones: {}".format(phones).encode('utf-8'))
            # Iterate through tokens, construct requests and send to Dingtalk
            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 not found"), ' '.join(["@"+i for i in phones.keys()]))
                    },
                    "at": {
                        "atMobiles": list(phones.keys()),
                        "isAtAll": False
                    }
                }

                #print("Sending request to URL: {}".format(url).encode('utf-8'))
                #print("Request body: {}".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("Response status code: {}".format(response.getcode()).encode('utf-8'))
                #print("Response text: {}".format(result).encode('utf-8'))
        # Catch all exceptions to prevent program crashes
        except Exception as e:
            #print("Error in send_newtalk: {}".format(e).encode('utf-8'))

def main():
    try:
        #print("Reading payload from stdin".encode('utf-8'))
        payload = json.load(sys.stdin)
        with open(".payload", 'w') as f:
            f.write(json.dumps(payload, indent=4))
        #print("Payload written to .payload file".encode('utf-8'))

        for ch in payload.get('event').get('notify_channels'):
            send_func_name = "send_{}".format(ch.strip())
            #print("Processing channel: {}, function name: {}".format(ch, send_func_name).encode('utf-8'))
            if not hasattr(Sender, send_func_name):
                #print("Function {} not found".format(send_func_name).encode('utf-8'))
                continue
            send_func = getattr(Sender, send_func_name)
            send_func(payload)

    except Exception as e:
        print("Error in main: {}".format(e).encode('utf-8'))

def hello():
    print("hello nightingale".encode('utf-8'))

if __name__ == "__main__":
    if len(sys.argv) == 1:
        main()
    elif sys.argv[1] == "hello":
        hello()
    else:
        print("I am confused".encode('utf-8'))

The sample script contains many print statements for output, mainly for debugging purposes. If enabled, these can be viewed through n9e logs. Debug logs are in INFO.log, error states are in ERROR.log, filter for the keyword event_script_notify

脚本日志

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