maubot-alertbot/alertbot.py
2022-12-19 15:29:46 +01:00

222 lines
7.6 KiB
Python

from maubot import Plugin, MessageEvent
from maubot.handlers import web, command
from aiohttp.web import Request, Response, json_response
import json
import datetime
from mautrix.errors.request import MForbidden
helpstring = f"""# Alertbot
To control the alertbot you can use the following commands:
* `!help`: To show this help
* `!ping`: To check if the bot is alive
* `!raw`: To toggle raw mode (where webhook data is not parsed but simply forwarded as copyable text)
* `!roomid`: To let the bot show you the current matrix room id
* `!url`: To let the bot show you the webhook url
More information is on [Github](https://github.com/moan0s/alertbot)
"""
def get_alert_type(data):
"""
Currently supported are ["grafana-alert", "grafana-resolved", "prometheus-alert", "not-found"]
:return: alert type
"""
# Uptime-kuma has heartbeat
try:
if data["heartbeat"]["status"] == 0:
return "uptime-kuma-alert"
elif data["heartbeat"]["status"] == 1:
return "uptime-kuma-resolved"
except KeyError:
pass
# Grafana
try:
data["alerts"][0]["labels"]["grafana_folder"]
if data['status'] == "firing":
return "grafana-alert"
else:
return "grafana-resolved"
except KeyError:
pass
# Prometheus
try:
if data["alerts"][0]["labels"]["job"]:
if data['status'] == "firing":
return "prometheus-alert"
else:
return "prometheus-resolved"
except KeyError:
pass
return "not-found"
def get_alert_messages(alert_data: dict, raw_mode=False) -> list:
"""
Returns a list of messages in markdown format
:param alert_data: The data send to the bot as dict
:param raw_mode: Toggles a mode where the data is not parsed but simply returned as code block in a message
:return: List of alert messages in markdown format
"""
alert_type = get_alert_type(alert_data)
if raw_mode or alert_type == "not-found":
return ["**Data received**\n```\n" + str(alert_data).strip("\n").strip() + "\n```"]
else:
try:
if alert_type == "grafana-alert":
messages = grafana_alert_to_markdown(alert_data)
elif alert_type == "grafana-resolved":
messages = grafana_alert_to_markdown(alert_data)
elif alert_type == "prometheus-alert":
messages = prometheus_alert_to_markdown(alert_data)
elif alert_type == "prometheus-resolved":
messages = prometheus_alert_to_markdown(alert_data)
elif alert_type == "uptime-kuma-alert":
messages = uptime_kuma_alert_to_markdown(alert_data)
elif alert_type == "uptime-kuma-resolved":
messages = uptime_kuma_resolved_to_markdown(alert_data)
except KeyError as e:
messages = ["**Data received**\n```\n" + str(alert_data).strip(
"\n").strip() + f"\n```\nThe data was detected as {alert_type} but was not in an expected format. If you want to help the development of this bot, file a bug report [here](https://github.com/moan0s/alertbot/issues)\n{e.with_traceback()}"]
return messages
def uptime_kuma_alert_to_markdown(alert_data: dict):
tags_readable = ", ".join([tag["name"] for tag in alert_data["monitor"]["tags"]])
message = (
f"""**Firing 🔥**: Monitor down: {alert_data["monitor"]["url"]}
* **Error:** {alert_data["heartbeat"]["msg"]}
* **Started at:** {alert_data["heartbeat"]["time"]}
* **Tags:** {tags_readable}
* **Source:** "Uptime Kuma"
"""
)
return [message]
def uptime_kuma_resolved_to_markdown(alert_data: dict):
tags_readable = ", ".join([tag["name"] for tag in alert_data["monitor"]["tags"]])
message = (
f"""**Resolved 💚**: {alert_data["monitor"]["url"]}
* **Status:** {alert_data["heartbeat"]["msg"]}
* **Started at:** {alert_data["heartbeat"]["time"]}
* Duration until resolved {alert_data["heartbeat"]["duration"]}s
* **Tags:** {tags_readable}
* **Source:** "Uptime Kuma"
"""
)
return [message]
def grafana_alert_to_markdown(alert_data: dict) -> list:
"""
Converts a grafana alert json to markdown
:param alert_data:
:return: Alerts as formatted markdown string list
"""
messages = []
for alert in alert_data["alerts"]:
datetime_format = "%Y-%m-%dT%H:%M:%S%z"
if alert['status'] == "firing":
message = (
f"""**Firing 🔥**: {alert['labels']['alertname']}
* **Instance:** {alert["valueString"]}
* **Silence:** {alert["silenceURL"]}
* **Started at:** {alert['startsAt']}
* **Fingerprint:** {alert['fingerprint']}
"""
)
if alert['status'] == "resolved":
end_at = datetime.datetime.strptime(alert['endsAt'], datetime_format)
start_at = datetime.datetime.strptime(alert['startsAt'], datetime_format)
message = (
f"""**Resolved 🥳**: {alert['labels']['alertname']}
* **Duration until resolved:** {end_at - start_at}
* **Fingerprint:** {alert['fingerprint']}
"""
)
messages.append(message)
return messages
def prometheus_alert_to_markdown(alert_data: dict) -> str:
"""
Converts a prometheus alert json to markdown
:param alert_data:
:return: Alert as fomatted markdown
"""
messages = []
for alert in alert_data["alerts"]:
message = (
f"""**{alert['status']}** {'💚' if alert['status'] == 'resolved' else '🔥'}: {alert['annotations']['description']}
* **Alertname:** {alert["labels"]['alertname']}
* **Instance:** {alert["labels"]['instance']}
* **Job:** {alert["labels"]['job']}
"""
)
messages.append(message)
return messages
class AlertBot(Plugin):
raw_mode = False
async def send_alert(self, req, room):
text = await req.text()
self.log.info(text)
content = json.loads(f"{text}")
for message in get_alert_messages(content, self.raw_mode):
self.log.debug(f"Sending alert to {room}")
await self.client.send_markdown(room, message)
@web.post("/webhook/{room_id}")
async def webhook_room(self, req: Request) -> Response:
room_id = req.match_info["room_id"].strip()
try:
await self.send_alert(req, room=room_id)
except MForbidden:
self.log.error(f"Could not send to {room_id}: Forbidden. Most likely the bot is not invited in the room.")
return json_response('{"status": "forbidden", "error": "forbidden"}', status=403)
return json_response({"status": "ok"})
@command.new()
async def ping(self, evt: MessageEvent) -> None:
"""Answers pong to check if the bot is running"""
await evt.reply("pong")
@command.new()
async def roomid(self, evt: MessageEvent) -> None:
"""Answers with the current room id"""
await evt.reply(f"`{evt.room_id}`")
@command.new()
async def url(self, evt: MessageEvent) -> None:
"""Answers with the url of the webhook"""
await evt.reply(f"`{self.webapp_url}/webhook`")
@command.new()
async def raw(self, evt: MessageEvent) -> None:
self.raw_mode = not self.raw_mode
"""Switches the bot to raw mode or disables raw mode (mode where data is not formatted but simply forwarded)"""
await evt.reply(f"Mode is now: `{'raw' if self.raw_mode else 'normal'} mode`")
@command.new()
async def help(self, evt: MessageEvent) -> None:
await self.client.send_markdown(evt.room_id, helpstring)