groups:
  - name: disk_usage_alerts
    rules:
      - alert: DiskUsageHigh
        expr: 100 * (node_filesystem_size_bytes{fstype!~"tmpfs|fuse.lxcfs|overlay"} - node_filesystem_avail_bytes{fstype!~"tmpfs|fuse.lxcfs|overlay"}) / node_filesystem_size_bytes{fstype!~"tmpfs|fuse.lxcfs|overlay"} > 40
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Disk usage is high on {{ $labels.instance }}"
          description: "Disk usage on {{ $labels.instance }} is above 40% (current value: {{ $value }}%)"

# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
rule_files:
  - "/data/prometheus/alert.rules.yml"

# Alertmanager configuration
alerting:
  alertmanagers:
    - static_configs:
        - targets: ["127.0.0.1:9093"]

route:
  group_by: ['disk_usage_alerts']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 1h
  receiver: 'webhook_receiver'
receivers:
  - name: 'webhook_receiver'
    webhook_configs:
            - url: 'http://127.0.0.1:4000/alert'

import requests
import json
import time
from datetime import datetime, timedelta

# Author: Empty Box
alert_manager_url = "http://127.0.0.1:9093/api/v2/alerts"
webhook_url = «https://open.feishu.cn/ваш собственный адрес робота Feishu»


# Получить оповещение
def get_alerts(alert_manager_url):
    try:
        response = requests.get(alert_manager_url)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching alerts: {e}")
        return []


# Мониторинг тревог
def monitor_alerts(alert_manager_url, poll_interval=60):
    known_alerts = set()

    while True:
        alerts = get_alerts(alert_manager_url)
        new_alerts = []

        for alert in alerts:
            alert_id = alert["labels"].get("alertname")
            if alert_id not in known_alerts:
                known_alerts.add(alert_id)
                new_alerts.append(alert)

        if new_alerts:
            print(f"New alerts found: {len(new_alerts)}")
            for alert in new_alerts:
                print(json.dumps(alert, indent=4))

        time.sleep(poll_interval)


# Отправить оповещение
def send_alert(alert_manager_url):
    try:
        response = requests.get(alert_manager_url)
        if response.status_code == 200:
            alert_messages = response.json()
            for alert_message in alert_messages:
                labels = alert_message['labels']
                # Тип тревоги
                alert_name = alert_message['labels']['alertname']
                # Тип задачи
                job = alert_message['labels']['job']
                # Уровень тревоги
                severity = alert_message['labels']['severity']
                # Содержимое тревоги
                description = alert_message['annotations']['description']
                # Получить оповещениесостояние,привык судитьда Повторять ли сигнал тревоги
                state = alert_message['status']['state']
                # Создайте тело сообщения о тревоге робота.
                webhook_msg = {}
                elements = []
                elements_json = {}
                alert_name_json = {}
                alert_name_text_json = {}
                description_json = {}
                description_text_json = {}
                job_json = {}
                job_text_json = {}

                msg_type = "interactive"
                webhook_msg["msg_type"] = msg_type

                alert_name_json["tag"] = "div"
                alert_name_text_json["content"] = "**Тип тревоги**：" + alert_name
                alert_name_text_json["tag"] = "lark_md"
                alert_name_json["text"] = alert_name_text_json
                elements.append(alert_name_json)

                start_time_json["tag"] = "div"
                start_time_text_json["content"] = "**Время срабатывания**:" + start_time
                start_time_text_json["tag"] = "lark_md"
                start_time_json["text"] = start_time_text_json
                elements.append(start_time_json)

                severity_json["tag"] = "div"
                severity_text_json["content"] = "**Уровень тревоги**：" + severity
                severity_text_json["tag"] = "lark_md"
                severity_json["text"] = severity_text_json
                elements.append(severity_json)

                instance_id_json["tag"] = "div"
                instance_id_text_json["content"] = "**Экземпляр триггера**:" + instance_id
                instance_id_text_json["tag"] = "lark_md"
                instance_id_json["text"] = instance_id_text_json
                elements.append(instance_id_json)

                job_json["tag"] = "div"
                job_text_json["content"] = "**Тип мониторинга**:" + job
                job_text_json["tag"] = "lark_md"
                job_json["text"] = job_text_json
                elements.append(job_json)

                description_json["tag"] = "div"
                description_text_json["content"] = "**Содержимое тревоги**：" + description
                description_text_json["tag"] = "lark_md"
                description_json["text"] = description_text_json
                elements.append(description_json)

                elements_json["elements"] = elements
                webhook_msg["card"] = elements_json
                json_data = json.dumps(webhook_msg)
                requests.post(webhook_url, headers={'Content-Type': 'application/json'}, data=json_data)

        else:
            print('Не удалось получить информацию о тревоге')
    except Exception as e:
        print(f"error is :{e}")


if __name__ == "__main__":
    poll_interval = 60  # Интервал опроса в секундах
    monitor_alerts(alert_manager_url, poll_interval)

from flask import Flask, request, jsonify
from datetime import datetime, timedelta
import json
import time
import requests

app = Flask(__name__)

webhook_url = «https://open.feishu.cn/ваш собственный адрес робота»


@app.route('/alert', methods=['POST'])
def receive_alert():
    # Вызывается Alert для анализа информации JSON, возвращаемой Alert.
    try:
        msg_json = {}
        alert_response = request.json
        alerts = alert_response["alerts"]
        if len(alerts) > 0:
            for alert in alerts:
                alert_name = alert["labels"]["alertname"]
                # По типу светильникосуществлять различные тела сообщений с информацией о тревогах
                if "DiskUsageHigh" == alert_name:
                    msg_json = disk_usage_high_json(alert)
                    # Отправить оповещение
                    response = send_alert(msg_json)
                    print("Исходные данные тревоги: ", alert)
                    continue

                return str(response.status_code)
    except Exception as e:
        return e


def send_alert(json_data):
    response = requests.post(webhook_url, headers={'Content-Type': 'application/json'}, data=json_data)
    return response


def disk_usage_high_json(alerts):
    alert_name = alerts["labels"]["alertname"]
    job = alerts["labels"]["job"]
    description = alerts["annotations"]["description"]
 

    webhook_msg = {}
    elements = []
    elements_json = {}
    alert_name_json = {}
    alert_name_text_json = {}
    description_json = {}
    description_text_json = {}
    job_json = {}
    job_text_json = {}

    msg_type = "interactive"
    webhook_msg["msg_type"] = msg_type

    alert_name_json["tag"] = "div"
    alert_name_text_json["content"] = "**Тип тревоги**：" + alert_name
    alert_name_text_json["tag"] = "lark_md"
    alert_name_json["text"] = alert_name_text_json
    elements.append(alert_name_json)

    start_time_json["tag"] = "div"
    start_time_text_json["content"] = "**Время срабатывания**:" + start_time
    start_time_text_json["tag"] = "lark_md"
    start_time_json["text"] = start_time_text_json
    elements.append(start_time_json)

    severity_json["tag"] = "div"
    severity_text_json["content"] = "**Уровень тревоги**：" + severity
    severity_text_json["tag"] = "lark_md"
    severity_json["text"] = severity_text_json
    elements.append(severity_json)

    description_json["tag"] = "div"
    description_text_json["content"] = "**Содержимое тревоги**：" + description
    description_text_json["tag"] = "lark_md"
    description_json["text"] = description_text_json
    elements.append(description_json)

    elements_json["elements"] = elements
    webhook_msg["card"] = elements_json
    json_data = json.dumps(webhook_msg)

    return json_data



if __name__ == '__main__':
    app.run(host='0.0.0.0', port=4000)