97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
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"<strong><font color={color}>{header_str}</font>: {title}</strong>"
|
|
)
|
|
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"<br/>Description: {description}"
|
|
plain += f" \n{description}"
|
|
|
|
formatted += f"<br/>{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 += "<br/>" + " | ".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)
|