Docs/docs/services/lxc208-manage/alert

Matrix Alert Relay — Документация

1. Архитектура и структура проекта

Проект представляет собой HTTP-сервер (вебхук-релей), который принимает POST-запросы от систем мониторинга, парсит их, форматирует и отправляет в нужные комнаты Matrix.

Основные файлы:

• matrix-relay.py — точка входа. HTTP-сервер, принимающий запросы и маршрутизирующий их.
• config.py — глобальные настройки (адрес Homeserver, токен, порт).
• rooms.py — маппинг логических имен каналов (disk, backup, docker) на ID комнат Matrix.
• alerts.py — словарь ALERTS_CONFIG с описанием всех известных типов алертов.
• grafana.py — обработчик стандартных вебхуков от Grafana/Prometheus.
• pbs.py — обработчик специфичных вебхуков от Proxmox VE / Proxmox Backup Server.
• formatters.py — функции красивого форматирования текста для специфичных алертов.
• matrix.py — низкоуровневая отправка HTML-сообщений в Matrix API.
• utils.py — вспомогательные утилиты (экранирование HTML, форматирование байтов и т.д.).

---

2. Сценарий А: Добавление нового алерта (Grafana / Prometheus)

Если новый алерт приходит из Grafana или Prometheus, писать новый код парсинга не нужно. Достаточно обновить конфигурацию.

Шаг 2.1. Добавление комнаты (если нужна новая)

Открой rooms.py и добавь новое имя канала и ID комнаты:

CHANNELS = {
    "default": "!WlzSAzmLQEATwDCaBU:matrix.zailon.ru",
    "disk": "!wOlscPjqPWVbSHpYgc:matrix.zailon.ru",
    "backup": "!ViUbvoFxTTbRLMtZEO:matrix.zailon.ru",
    "ups": "!NewRoomId:matrix.zailon.ru",
}

Шаг 2.2. Создание форматтера (опционально)

Если алерт требует особого вывода (например, вывод заряда батареи в процентах), открой formatters.py:

def ups_battery(device, value=None):
    if value is not None:
        return f"🔋 {device:<8} {value}%"
    return f"🔋 {device:<8}"

FORMATTERS = {
    "temperature": temperature,
    "ups_battery": ups_battery,
}

Шаг 2.3. Регистрация алерта

Открой alerts.py и добавь новый ключ в ALERTS_CONFIG. Имя ключа должно точно совпадать с alertname из Grafana.

ALERTS_CONFIG = {
    "UPS Battery Low": {
        "room": "ups",
        "title": "🔋 Низкий заряд ИБП",
        "summary": "⚡ Заряд батареи ИБП упал ниже критического.",
        "info": "🔌 Проверьте питание и состояние ИБП.",
        "resolved": "✅ Питание ИБП восстановлено.",
        "formatter": "ups_battery",
        "skip_zero_firing": False,
    },
}

Релей автоматически подхватит новый алерт, отформатирует его и отправит в нужную комнату.

---

3. Сценарий Б: Добавление нового ИСТОЧНИКА алертов (кастомный вебхук)

Если новая система (например, Docker Events, Zabbix, или скрипт мониторинга UPS) шлет свой собственный JSON, не похожий на формат Grafana, нужно написать новый модуль.

Шаг 3.1. Создание модуля-обработчика

Создай новый файл, например ups.py (по аналогии с pbs.py).

import logging
from alerts.matrix import send_matrix

logger = logging.getLogger(__name__)

def process_ups_alert(obj: dict) -> None:
    ups_name = obj.get("ups_name", "Unknown UPS")
    battery_level = obj.get("battery", 0)
    status = obj.get("status", "unknown")
    
    if status == "critical":
        text = f"<b>🚨 Критический разряд ИБП</b>\n\n"
        text += f"🔋 <b>{ups_name}</b>\n"
        text += f"📉 Осталось заряда: <b>{battery_level}%</b>\n"
        text += "⚡ Срочно проверьте питание!"
        room = "ups"
    else:
        text = f"<b>ℹ️ Статус ИБП</b>\n\n{ups_name}: {battery_level}%"
        room = "default"
        
    send_matrix(text, room)

Шаг 3.2. Подключение обработчика в роутер

Открой matrix-relay.py. В методе process_webhook добавь условие для определения нового источника и вызови свою функцию.

from alerts.ups import process_ups_alert

class WebhookHandler(BaseHTTPRequestHandler):
    def process_webhook(self, obj: Dict) -> None:
        if "ups_name" in obj:
            logger.info("Processing UPS notification")
            process_ups_alert(obj)
            return

        source = detect_backup_source(obj)
        if source:
            logger.info("Processing backup notification (source=%s)", source.value)
            process_backup(obj, send_matrix, source)
            return

        if "alerts" in obj:
            alerts = obj["alerts"]
            logger.info("Processing Grafana alerts: %d items", len(alerts))
            process_grafana({"alerts": alerts}, send_matrix)
            return

        logger.warning("Unknown webhook type")
        send_matrix(
            "<b>Webhook received</b><br><br><pre>" +
            json.dumps(obj, ensure_ascii=False, indent=2) +
            "</pre>"
        )

Теперь при отправке POST-запроса на сервер с JSON {"ups_name": "APC", "battery": 10, "status": "critical"} алерт улетит в Matrix.

---

4. Шпаргалка по настройке вебхуков

• URL для всех источников: http://<IP-сервера-manage>:8086/ (или https://, если стоит реверс-прокси).
• Метод: POST
• Content-Type: application/json

Пример вызова из bash-скрипта (для Docker, UPS, Cron):

curl -X POST http://127.0.0.1:8086/ \
     -H "Content-Type: application/json" \
     -d '{"ups_name": "ServerRoom_UPS", "battery": 5, "status": "critical"}'

---

5. Логирование и отладка

Все события пишутся в стандартный вывод. Если релей запущен через systemd, смотри логи:

journalctl -u matrix-relay -f

Если алерт пришел, но не обработался так, как ожидалось, ищи в логах строки:

• Processing Grafana alerts — релей распознал формат Grafana.
• Processing backup notification — релей распознал бэкап PVE/PBS.
• Unknown webhook type — релей не нашел знакомых ключей и отправил "сырой" JSON в комнату default. Используй это для отладки новых источников!