Добавить docs/services/lxc208-manage/alert/grafana.py

docs/services/lxc208-manage/alert/grafana.py

@@ -0,0 +1,230 @@
+import re
+import logging
+
+from alerts.alerts import ALERTS_CONFIG
+from alerts.utils import escape_html, safe_int
+from alerts.formatters import get_formatter
+
+logger = logging.getLogger(__name__)
+
+
+def is_backup_alert(alert):
+    labels = alert.get("labels") or {}
+    alertname = labels.get("alertname", "")
+    if "backup" in alertname.lower():
+        return True
+    if "backup" in labels:
+        return True
+    return False
+
+
+def build_backup_alerts(alerts, config, alertname):
+    title = config["title"]
+    firing_backups = []
+    resolved_backups = []
+
+    for alert in alerts:
+        status = alert.get("status", "unknown")
+        if status not in ("firing", "resolved"):
+            continue
+        labels = alert.get("labels") or {}
+        backup_name = labels.get("backup", "unknown")
+
+        if status == "firing":
+            firing_backups.append(backup_name)
+        else:
+            resolved_backups.append(backup_name)
+
+    if not firing_backups and not resolved_backups:
+        return None
+
+    lines = []
+
+    if not firing_backups and resolved_backups:
+        lines.append(f"<b>{config.get('resolved', title)}</b>")
+    else:
+        lines.append(f"<b>{title}</b>")
+
+    lines.append("")
+
+    if firing_backups:
+        lines.append("\U0001F525 \u0423\u0441\u0442\u0430\u0440\u0435\u043B\u0438:")
+        for b in sorted(firing_backups):
+            lines.append(f"  \U0001F4E6 {escape_html(b)}")
+        lines.append("")
+
+    if resolved_backups:
+        lines.append("\u2705 \u041e\u0431\u043d\u043e\u0432\u0438\u043b\u0438\u0441\u044c:")
+        for b in sorted(resolved_backups):
+            lines.append(f"  \U0001F4E6 {escape_html(b)}")
+        lines.append("")
+
+    if firing_backups:
+        if "summary" in config:
+            lines.append(config['summary'])
+        if "info" in config:
+            lines.append(config['info'])
+
+    return "\n".join(lines)
+
+
+def collect_devices(alerts, config):
+    formatter_name = config.get("formatter")
+    format_value = get_formatter(formatter_name) if formatter_name else None
+    skip_zero = config.get("skip_zero_firing", False)
+
+    devices_list = []
+    firing_count = 0
+    resolved_count = 0
+
+    for alert in alerts:
+        status = alert.get("status", "unknown")
+        if status not in ("firing", "resolved"):
+            continue
+        alert_labels = alert.get("labels") or {}
+        values = alert.get("values") or {}
+        device = alert_labels.get("device") or alert_labels.get("disk") or alert_labels.get("instance", "")
+
+        if "A" in values:
+            value = safe_int(values["A"])
+            if status == "firing" and skip_zero and value is not None and value == 0:
+                continue
+            if value is not None and format_value:
+                line = format_value(device, value)
+            else:
+                line = f"{device:<8}"
+        elif device:
+            if format_value:
+                line = format_value(device)
+            else:
+                line = f"{device:<8}"
+        else:
+            continue
+
+        if status == "resolved":
+            line += " \u2705"
+            resolved_count += 1
+        else:
+            firing_count += 1
+
+        devices_list.append(line)
+
+    return devices_list, firing_count, resolved_count
+
+
+def build_header(config, firing_count, resolved_count):
+    title = config["title"]
+    if firing_count == 0 and resolved_count > 0:
+        return config.get('resolved', title)
+    return title
+
+
+def build_footer(config, firing_count):
+    lines = []
+    if firing_count > 0:
+        if "summary" in config:
+            lines.append(config['summary'])
+        if "info" in config:
+            lines.append(config['info'])
+    return lines
+
+
+def build_known_alert(alerts, config, alertname):
+    devices_list, firing_count, resolved_count = collect_devices(alerts, config)
+
+    if firing_count == 0 and resolved_count == 0:
+        return None
+
+    lines = []
+    lines.append(f"<b>{build_header(config, firing_count, resolved_count)}</b>")
+    lines.append("")
+
+    if devices_list:
+        lines.append("\n".join(devices_list))
+        lines.append("")
+
+    footer = build_footer(config, firing_count)
+    if footer:
+        lines.extend(footer)
+
+    return "\n".join(lines)
+
+
+def build_unknown_alert(alerts):
+    lines = []
+
+    if any(a.get("status") == "firing" for a in alerts):
+        lines.append("\U0001F6A8 Grafana Alert")
+    else:
+        lines.append("\u2705 Grafana Recovery")
+
+    lines.append("")
+
+    for alert in alerts:
+        status = alert.get("status", "unknown")
+        if status not in ("firing", "resolved"):
+            continue
+        alert_labels = alert.get("labels") or {}
+        annotations = alert.get("annotations") or {}
+        values = alert.get("values") or {}
+        alert_name = alert_labels.get("alertname", "Alert")
+        device = alert_labels.get("device") or alert_labels.get("disk") or alert_labels.get("instance", "")
+        summary = annotations.get("summary", "")
+        description = annotations.get("description", "") or ""
+        icon = "\U0001F525" if status == "firing" else "\u2705"
+
+        lines.append(f"{icon} {escape_html(alert_name)}")
+        if device:
+            lines.append(f"\U0001F4BD {escape_html(device)}")
+        if "A" in values:
+            value = safe_int(values["A"])
+            if value is not None:
+                if "temp" in alert_name.lower():
+                    lines.append(f"\U0001F321 {value}\u00b0C")
+                elif "bad block" in alert_name.lower() or "defect" in alert_name.lower():
+                    lines.append(f"\U0001F525 Bad blocks: {value}")
+                else:
+                    lines.append(f"\U0001F4C8 {value}")
+        if summary:
+            lines.append(f"\U0001F4DD {escape_html(summary)}")
+        if description:
+            clean_desc = re.sub(r"\[ var=.*?\]", "", description)
+            clean_desc = re.sub(r"\{[^}]*\}", "", clean_desc)
+            clean_desc = " ".join(clean_desc.split())
+            if clean_desc and len(clean_desc) > 5:
+                lines.append(f"\u2139 {escape_html(clean_desc)}")
+        lines.append("<hr>")
+
+    return "\n".join(lines)
+
+
+def process_grafana(obj, send_func):
+    alerts = obj.get("alerts", [])
+    if not alerts:
+        send_func("<b>Webhook received</b><br><br><pre>" + escape_html(
+            __import__("json").dumps(obj, ensure_ascii=False, indent=2)
+        ) + "</pre>")
+        return
+
+    groups = {}
+    for alert in alerts:
+        labels = alert.get("labels") or {}
+        name = labels.get("alertname", "Unknown")
+        groups.setdefault(name, []).append(alert)
+
+    for name, group in groups.items():
+        config = ALERTS_CONFIG.get(name)
+
+        if config:
+            channel = config.get("room", "default")
+
+            if is_backup_alert(group[0]):
+                text = build_backup_alerts(group, config, name)
+            else:
+                text = build_known_alert(group, config, name)
+        else:
+            channel = "default"
+            text = build_unknown_alert(group)
+
+        if text:
+            send_func(text, channel)