告警 API 调用示例

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import base64
import requests
import json

data = {
    "alarmType": "your_alarm_type", 
    "labels": {"key":"value"}, 
    "target":"alarm_target",
}

base64userpass = base64.b64encode('{0}:{1}'.format('username', 'password'))
headers = {
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    "Authorization": "Basic %s" % base64userpass,
}

resp = requests.post(url='http://192.168.0.1:8080/api/v2/alarm/alarms', headers=headers, data=json.dumps(data))
jresp = json.loads(resp.text)
print(jresp)

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# 注意：脚本内容第一行必须是 shebang 指定执行的程序，只支持 python 和 bash

import json
import requests
import os
import sys

ACCESS_TOKEN = "Ding Talk Token"

def send_alarm():
    """
    可以通过 os.environ 从环境变量获取 告警相关的变量值。
    支持的变量列表参见 《用户指南/附录4 OCP 告警模板变量》。
    """
    message = os.environ['message']
    data = {
        "msgtype": "markdown",
        "markdown": {
            "text": message
        }
    }
    resp = requests.post("https://oapi.dingtalk.com/robot/send?access_token=" + ACCESS_TOKEN, json=data)
    resp.close()
    response = json.loads(resp.text)

    # 将返回结果输入到标准错误（stderr）或标准输出（stdout），会被用于验证告警是否发送成功，优先验证stderr。
    if not response['errcode'] == 0:
        sys.stderr.write(resp.text)
    else:
        print(resp.text)

def main():
    """
    若发送成功，返回值为 0
    若发送失败，错误输出到 stderr，返回值不为 0
    """
    try:
        send_alarm()
        return 0
    except Exception as e:
        sys.stderr.write(str(e))
        return 1

if __name__ == "__main__":
    sys.exit(main())