创建群聊

添加机器人

配置安全设置为加签，并记录Webhook和加签密钥

编写告警脚本

在alertscripts目录下添加以下脚本，修改Webhook与加签密钥

#!/usr/bin/python3
# coding:utf-8 
​
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json
import sys
​
​
api_url = 'https://oapi.dingtalk.com/robot/send?access_token=' \
              'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx'   # webhook token粘贴在这里
​
​
# 计算签名
def get_timestamp_sign():
    """
    official url https://open.dingtalk.com/document/robots/customize-robot-security-settings
    :return: timestamp, sign
    """
    timestamp = str(round(time.time() * 1000))
    secret = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'  # 加签密钥粘贴在这里
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign
​
​
# 获取签名计算后的链接
def get_signed_url():
    """
    :拼接url format --> https://oapi.dingtalk.com/robot/send?access_token=XXXXXX&timestamp=XXX&sign=XXX
    :return:  webhook
    """
    timestamp, sign = get_timestamp_sign()
    webhook = api_url + "&timestamp=" + timestamp + "&sign=" + sign
    return webhook
​
​
# 定义webhook消息模式
def get_webhook(mode):
    if mode == 0:                 # 仅关键字
        webhook = api_url
    elif mode == 1 or mode == 2:  # 关键字+加签 或 关键字+加签+ip
        webhook = get_signed_url()
    else:
        webhook = ""
        print("error! mode:   ", mode, "  webhook :  ", webhook)
    print(webhook)
    return webhook
​
​
def get_message(text, user_info):
    message = {
        "msgtype": "text",  # 有text, "markdown"、link、整体跳转ActionCard 、独立跳转ActionCard、FeedCard类型等
        "text": {
            "content": text  # 消息内容
        },
        "at": {
            "atMobiles": [
                user_info,
            ],
            "isAtAll": False  # 是否是发送群中全体成员
        }
    }
    return message
​
​
def send_ding_message(text, user_info):
    """
    model: 0 --> 关键字; 1 --> 关键字 +加签; 2 --> 关键字+加签+IP
    :param text:
    :param user_info:
    :return:
    """
    webhook = get_webhook(1)
    # 构建请求头部
    header = {
        "Content-Type": "application/json",
        "Charset": "UTF-8"
    }
    message = get_message(text, user_info)
    message_json = json.dumps(message)
    info = requests.post(url=webhook, data=message_json, headers=header).json()
    code = info["errcode"]
    errmsg = info["errmsg"]
    now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    if code == 0:
        print(now_time, ':Message sent successfully, return info:{} {}\n'.format(code, errmsg))
    else:
        print(now_time + ":Message sending failed, return info:{} {}\n".format(code, errmsg))
        exit(3)
​
​
if __name__ == "__main__":
    t = sys.argv[3]
    u = sys.argv[1]
    send_ding_message(t, u)

测试发送信息

./dingding.py all subject "test message"

添加告警媒介

添加动作

触发器触发后告警给用户群组

添加用户告警媒介，收件人为钉钉注册的手机号，如果发送的是具体人会@。

配置告警消息

当前状态: {TRIGGER.STATUS}
告警主机: {HOST.NAME}
告警 IP: {HOST.IP}
告警时间: {EVENT.DATE}-{EVENT.TIME}
告警等级: {TRIGGER.SEVERITY}
告警信息: {TRIGGER.NAME}
告警项目: {TRIGGER.KEY1}

测试触发告警

占用cpu单核心负载，根据核数开对应的终端，每个终端运行以下命令。

while : ; do  openssl speed; done

当前是4核主机，开四个线程