121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Alertmanager → Telegram bridge.
|
|
Receives Alertmanager webhooks at /alert and forwards formatted messages
|
|
to Telegram via Bot API. Also has /health endpoint for monitoring.
|
|
|
|
Environment variables:
|
|
TELEGRAM_BOT_TOKEN — Bot API token
|
|
TELEGRAM_CHAT_ID — Target chat ID (user or group)
|
|
|
|
Deploy via Docker: see templates/monitoring-docker-compose.yml
|
|
Dockerfile: FROM python:3.12-slim; pip install flask requests; CMD python bridge.py
|
|
"""
|
|
import json
|
|
import os
|
|
import requests
|
|
from flask import Flask, request, jsonify
|
|
|
|
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
|
|
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
def send_telegram(message: str):
|
|
"""Send a message via Telegram Bot API."""
|
|
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
|
payload = {
|
|
"chat_id": CHAT_ID,
|
|
"text": message,
|
|
"parse_mode": "Markdown",
|
|
"disable_web_page_preview": True,
|
|
}
|
|
try:
|
|
resp = requests.post(url, json=payload, timeout=10)
|
|
return resp.status_code == 200
|
|
except Exception as e:
|
|
print(f"Telegram send error: {e}")
|
|
return False
|
|
|
|
|
|
def format_alert(alert: dict) -> str:
|
|
"""Format a single Alertmanager alert as a Markdown message."""
|
|
status = alert.get("status", "unknown")
|
|
labels = alert.get("labels", {})
|
|
annotations = alert.get("annotations", {})
|
|
|
|
icon = "🔴" if status == "firing" else "🟢"
|
|
|
|
lines = [
|
|
f"{icon} *{labels.get('alertname', 'Unknown')}* — `{labels.get('instance', '')}`",
|
|
f"*Severity:* {labels.get('severity', '?')}",
|
|
f"*Description:* {annotations.get('description', 'N/A')}",
|
|
]
|
|
if annotations.get("summary"):
|
|
lines.append(f"*Summary:* {annotations['summary']}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
@app.route("/alert", methods=["POST"])
|
|
def receive_alert():
|
|
"""Receive Alertmanager webhook and forward to Telegram."""
|
|
data = request.get_json(force=True, silent=True)
|
|
if not data:
|
|
return jsonify({"error": "invalid JSON"}), 400
|
|
|
|
alerts = data.get("alerts", [])
|
|
if not alerts:
|
|
return jsonify({"status": "no alerts"}), 200
|
|
|
|
for alert in alerts:
|
|
message = format_alert(alert)
|
|
send_telegram(message)
|
|
|
|
return jsonify({"status": "sent", "count": len(alerts)}), 200
|
|
|
|
|
|
@app.route("/pve", methods=["POST"])
|
|
def pve_notification():
|
|
"""Handle PVE webhook notifications (different format from Alertmanager).
|
|
|
|
PVE sends POST with JSON containing title/message/severity/type fields.
|
|
Register in PVE via:
|
|
pvesh create /cluster/notifications/endpoints/webhook \
|
|
--name telegram-bridge \
|
|
--url http://<monitoring-ct-ip>:9099/pve \
|
|
--method POST
|
|
Then test with:
|
|
pvesh create /cluster/notifications/test telegram-bridge
|
|
"""
|
|
data = request.get_json(silent=True)
|
|
if not data:
|
|
data = request.form.to_dict()
|
|
if not data:
|
|
return jsonify({"error": "no data"}), 400
|
|
|
|
title = data.get("title", data.get("subject", ""))
|
|
message = data.get("message", data.get("body", data.get("text", "")))
|
|
severity = data.get("severity", data.get("priority", "info"))
|
|
type_ = data.get("type", data.get("event", "notification"))
|
|
|
|
emoji = {"critical": "🔴", "warning": "🟡", "info": "🔵"}.get(severity, "🔵")
|
|
msg = f"{emoji} *PVE: {title or type_}*"
|
|
if message:
|
|
if len(message) > 3500:
|
|
message = message[:3500] + "\n...(truncated)"
|
|
msg += f"\n\n{message}"
|
|
|
|
ok = send_telegram(msg)
|
|
return jsonify({"status": "sent" if ok else "error"})
|
|
|
|
|
|
@app.route("/health")
|
|
def health():
|
|
return jsonify({"status": "ok"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=9099)
|