From 3d5f3c6299ecc7000692455c7614096364327114 Mon Sep 17 00:00:00 2001 From: Maxime Killinger Date: Mon, 31 Aug 2026 01:04:19 +0200 Subject: [PATCH] feat: align Alertmanager notifications with the CrowdSec feed format Alertmanager notifications were hard to read: the alert name appeared twice, the instance was repeated between the header and the summary, the count was restated in the body, and details ran flat with no visual structure. They now follow the same grammar as the CrowdSec feed: a title carrying the counts, then one block per alert made of an identity line (" ()") and details indented under a tree marker. Severity moves into the icon (red/orange/blue, green once resolved) instead of a "[critical]" suffix, which keeps the identity line short enough to read at a glance on a phone. --- app.py | 84 +++++++++++++++++++++++++++++++++------------------------- 1 file changed, 48 insertions(+), 36 deletions(-) diff --git a/app.py b/app.py index dd38041..4c95141 100644 --- a/app.py +++ b/app.py @@ -61,6 +61,43 @@ def format_scenario(scenario): return "unknown" return scenario.replace("crowdsecurity/", "").replace("_", " ") +# One icon per alert, mirroring the per-IP icon used by the CrowdSec feed. +# Severity is carried by the icon instead of a "[critical]" suffix, which +# keeps the identity line short enough to read at a glance on a phone. +SEVERITY_ICONS = { + "critical": "🔴", + "warning": "🟠", + "info": "🔵", +} + +def format_alert_block(alert, resolved=False): + """Render one alert the way the CrowdSec feed renders one IP: an identity + line (icon, name, context in parentheses), then the details indented under + a tree marker. Returns the block as a single string.""" + labels = alert.get("labels", {}) + annotations = alert.get("annotations", {}) + + name = labels.get("alertname", "alerte") + severity = labels.get("severity", "") + # "name" is what cAdvisor-based rules use to carry the container name + context = (labels.get("instance") or labels.get("container") + or labels.get("name") or labels.get("job") or "") + + icon = "✅" if resolved else SEVERITY_ICONS.get(severity, "🔥") + head = f"{icon} {name}" + if context: + head += f" ({context})" + + summary = annotations.get("summary") + description = annotations.get("description") + details = [summary] if summary else [] + # description often just restates summary; only keep it when it adds something + if description and description != summary: + details.append(description) + + lines = [head] + [f" └ {d}" for d in details] + return "\n".join(lines) + @app.route('/crowdsec', methods=['POST']) def handle_crowdsec(): try: @@ -120,48 +157,23 @@ def handle_alertmanager(): logger.debug("Received empty Alertmanager payload") return jsonify({"status": "no data"}), 200 - def format_alert(alert): - labels = alert.get("labels", {}) - annotations = alert.get("annotations", {}) - name = labels.get("alertname", "alerte") - severity = labels.get("severity", "") - instance = labels.get("instance") or labels.get("container") or labels.get("job") or "" - - head = name - if severity: - head += f" [{severity}]" - if instance: - head += f" — {instance}" - - out = [head] - summary = annotations.get("summary") - description = annotations.get("description") - if summary: - out.append(f" {summary}") - if description and description != summary: - out.append(f" {description}") - return "\n".join(out) - firing = [a for a in alerts if a.get("status") == "firing"] resolved = [a for a in alerts if a.get("status") == "resolved"] - lines = [] + # One block per alert, blank-line separated, like the CrowdSec feed + blocks = [format_alert_block(a) for a in firing] + blocks += [format_alert_block(a, resolved=True) for a in resolved] + + # Title carries the counts, so the body never has to repeat them + counts = [] if firing: - lines.append(f"🔥 {len(firing)} alerte(s) active(s)") - lines.extend(format_alert(a) for a in firing) + plural = "s" if len(firing) > 1 else "" + counts.append(f"{len(firing)} alerte{plural} active{plural}") if resolved: - if lines: - lines.append("") - lines.append(f"✅ {len(resolved)} résolue(s)") - lines.extend(format_alert(a) for a in resolved) + counts.append(f"{len(resolved)} résolue{'s' if len(resolved) > 1 else ''}") - status = data.get("status", "firing") - group_name = data.get("groupLabels", {}).get("alertname", "") - title = ("🔥" if status == "firing" else "✅") + " Alertmanager" - if group_name: - title += f" — {group_name}" - - body = "\n".join(lines).strip() + title = ("🔥" if firing else "✅") + f" Alertmanager - {', '.join(counts)}" + body = "\n\n".join(blocks) # Send to Apprise response = requests.post(APPRISE_URL, json={"title": title, "body": body}, timeout=10)