import logging from typing import Any, Dict, List, Tuple from fastapi import Request from ops_bot.common import ( COLOR_ALARM, COLOR_INFO, COLOR_OK, COLOR_UNKNOWN, COLOR_WARNING, ) from ops_bot.config import RoutingKey severity_colors = { "warning": COLOR_WARNING, "info": COLOR_INFO, "critical": COLOR_ALARM, } def prometheus_alert_to_markdown( alert_data: Dict, # type: ignore[type-arg] ) -> List[Tuple[str, str]]: """ Converts a prometheus alert json to markdown """ messages = [] ignore_labels = ["grafana_folder", "ref_id", "rulename", "datasource_uid"] logging.debug(f"alertmanager payload: {alert_data}") externalURL = alert_data.get("externalURL") for alert in alert_data["alerts"]: alertname = alert["labels"]["alertname"] title = alert["labels"]["alertname"] summary = alert["annotations"].get("summary") description = alert["annotations"].get("description") logging.debug(f"processing alert: '{title}'") labels = alert.get("labels", {}) severity = labels.get("severity", "unknown") status = alert.get("status", "unknown-status") if status == "resolved": color = COLOR_OK else: color = severity_colors.get(severity, COLOR_UNKNOWN) plain = f"{status.upper()}[{severity}]: {title}" header_str = f"{status.upper()}[{severity}]" formatted = ( f"{header_str}: {title}" ) if summary: formatted += f" - {summary}" plain += f" - {summary}" label_strings = [] for label_name, label_value in labels.items(): logging.debug(f"got label: {label_name} = {label_value}") if label_name.startswith("__"): continue if label_name in ignore_labels: continue label_strings.append((f"**{label_name.capitalize()}**: {label_value}")) labels_final = ", ".join(label_strings) plain += f" \n{labels_final}" if description: formatted += f"
Description: {description}" plain += f" \n{description}" formatted += f"
{labels_final}" generatorURL = alert.get("generatorURL") runbookURL = alert["annotations"].get("runbook_url") dashboardURL = alert.get("dashboardURL") silenceURL = alert.get("silenceURL") actions = [] if not silenceURL and externalURL: silence_filter = f"alertname%3D%22{alertname}%22" silence_filter = "{" + silence_filter + "}" silenceURL = f"{externalURL}/#/silences/new?filter={silence_filter}" if silenceURL: actions.append(f"[🔕 Mute]({silenceURL})") if runbookURL: actions.append(f"[📗 Runbook]({runbookURL})") if dashboardURL: actions.append(f"[📈 Dashboard]({dashboardURL})") if generatorURL: actions.append(f"[🔍 Inspect]({generatorURL})") formatted += "
" + " | ".join(actions) messages.append((plain, formatted)) return messages async def parse_alertmanager_event( route: RoutingKey, payload: Any, request: Request, ) -> List[Tuple[str, str]]: return prometheus_alert_to_markdown(payload)