230 lines
7.0 KiB
Python
230 lines
7.0 KiB
Python
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) |