日志易告警集成

# -*- coding: utf-8 -*-
import json

import requests

# Flashduty 集成推送地址，替换为控制台上「日志易」集成的推送地址
push_url = "https://api.flashcat.cloud/event/push/alert/standard?integration_key=YOUR_INTEGRATION_KEY"

# 前置 HTTP 代理，适用于日志易主机无法直连互联网的场景，不涉及则留空
proxies = {
    'http': '',
    'https': '',
}

# 日志易告警等级到 Flashduty 告警等级的映射
LEVEL_MAP = {
    "critical": "Critical",
    "high": "Critical",
    "mid": "Warning",
    "low": "Info",
    "info": "Info",
}

META = {
    "name": "FlashdutyWebhook",
    "version": 1,
    "alias": "Flashduty告警推送",
    "configs": []
}


def set_logger(reset_logger):
    global logger
    logger = reset_logger


def content(params, alert):
    return alert.get("description", "")


def handle(params, alert):
    level = alert.get("strategy", {}).get("trigger", {}).get("level", "low")
    if alert.get("is_alert_recovery"):
        event_status = "Ok"
    else:
        event_status = LEVEL_MAP.get(level, "Info")
    event = {
        "event_status": event_status,
        "title_rule": alert.get("name", "日志易告警"),
        "alert_key": "rizhiyi::" + alert.get("name", ""),
        "description": alert.get("description", ""),
        "labels": {
            "alert_name": alert.get("name", ""),
            "alert_level": level,
        },
    }
    resp = requests.post(push_url, json=event, timeout=10, proxies=proxies)
    logger.info("push to flashduty, status: %s, resp: %s", resp.status_code, resp.text)


def execute_reply(params, alert):
    handle(params, alert)
    return "push to flashduty done"