77 lines
2.3 KiB
Python
77 lines
2.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"]
|
|
|
|
logging.debug(f"alertmanager payload: {alert_data}")
|
|
for alert in alert_data["alerts"]:
|
|
title = (
|
|
alert["annotations"]["description"]
|
|
if hasattr(alert["annotations"], "description")
|
|
else alert["annotations"]["summary"]
|
|
)
|
|
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}]"
|
|
url = alert.get("generatorURL")
|
|
if url and url != "":
|
|
title_linked = f"[{title}]({url})"
|
|
else:
|
|
title_linked = title
|
|
formatted = (
|
|
f"<strong><font color={color}>{header_str}</font></strong>: {title_linked}"
|
|
)
|
|
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}"
|
|
formatted += f"<br/>{labels_final}"
|
|
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)
|