210 lines
7.2 KiB
Python
210 lines
7.2 KiB
Python
import logging
|
|
from typing import List
|
|
|
|
from diskcache import Cache
|
|
from nio import AsyncClient, MatrixRoom, RoomMessageText
|
|
|
|
from matrix_alertbot.alertmanager import AlertmanagerClient
|
|
from matrix_alertbot.chat_functions import react_to_event, send_text_to_room
|
|
from matrix_alertbot.config import Config
|
|
from matrix_alertbot.errors import AlertmanagerError
|
|
from matrix_alertbot.matcher import (
|
|
AbstractAlertMatcher,
|
|
AlertMatcher,
|
|
AlertRegexMatcher,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Command:
|
|
def __init__(
|
|
self,
|
|
client: AsyncClient,
|
|
cache: Cache,
|
|
alertmanager: AlertmanagerClient,
|
|
config: Config,
|
|
command: str,
|
|
room: MatrixRoom,
|
|
event: RoomMessageText,
|
|
):
|
|
"""A command made by a user.
|
|
|
|
Args:
|
|
client: The client to communicate to matrix with.
|
|
|
|
cache: Bot cache.
|
|
|
|
config: Bot configuration parameters.
|
|
|
|
command: The command and arguments.
|
|
|
|
room: The room the command was sent in.
|
|
|
|
event: The event describing the command.
|
|
"""
|
|
self.client = client
|
|
self.cache = cache
|
|
self.alertmanager = alertmanager
|
|
self.config = config
|
|
self.command = command
|
|
self.room = room
|
|
self.event = event
|
|
self.args = self.command.split()[1:]
|
|
|
|
async def process(self) -> None:
|
|
"""Process the command"""
|
|
if self.command.startswith("ack"):
|
|
await self._ack()
|
|
elif self.command.startswith("unack") or self.command.startswith("nack"):
|
|
await self._unack()
|
|
elif self.command.startswith("react"):
|
|
await self._react()
|
|
elif self.command.startswith("help"):
|
|
await self._show_help()
|
|
else:
|
|
await self._unknown_command()
|
|
|
|
async def _ack(self) -> None:
|
|
"""Acknowledge an alert and silence it for a certain duration in Alertmanager"""
|
|
matchers: List[AbstractAlertMatcher] = []
|
|
durations = []
|
|
for arg in self.args:
|
|
if "=~" in arg:
|
|
label, regex = arg.split("=~")
|
|
regex_matcher = AlertRegexMatcher(label, regex)
|
|
matchers.append(regex_matcher)
|
|
elif "=" in arg:
|
|
label, value = arg.split("=")
|
|
matcher = AlertMatcher(label, value)
|
|
matchers.append(matcher)
|
|
else:
|
|
durations.append(arg)
|
|
|
|
if len(durations) > 0:
|
|
duration = " ".join(durations)
|
|
else:
|
|
duration = "1d"
|
|
|
|
logger.debug(
|
|
f"Receiving a command to create a silence for a duration of {duration} | "
|
|
f"{self.room.user_name(self.event.sender)}: {self.event.body}"
|
|
)
|
|
|
|
source_content = self.event.source["content"]
|
|
try:
|
|
alert_event_id = source_content["m.relates_to"]["m.in_reply_to"]["event_id"]
|
|
except KeyError:
|
|
logger.debug("Unable to find the event ID of the alert")
|
|
return
|
|
logger.debug(f"Read alert fingerprints for event {alert_event_id} from cache")
|
|
|
|
count_created_silences = 0
|
|
alert_fingerprints = self.cache[alert_event_id]
|
|
for alert_fingerprint in alert_fingerprints:
|
|
logger.debug(
|
|
f"Create silence for alert with fingerprint {alert_fingerprint} for a duration of {duration}"
|
|
)
|
|
try:
|
|
await self.alertmanager.create_silence(
|
|
alert_fingerprint,
|
|
duration,
|
|
self.room.user_name(self.event.sender),
|
|
matchers,
|
|
)
|
|
count_created_silences += 1
|
|
except AlertmanagerError as e:
|
|
logger.exception(f"Unable to create silence: {e}", exc_info=e)
|
|
|
|
await send_text_to_room(
|
|
self.client,
|
|
self.room.room_id,
|
|
f"Created {count_created_silences} silences with a duration of {duration}.",
|
|
)
|
|
|
|
async def _unack(self) -> None:
|
|
"""Delete an alert's acknowledgement of an alert and remove corresponding silence in Alertmanager"""
|
|
logger.debug(
|
|
f"Receiving a command to delete a silence | "
|
|
f"{self.room.user_name(self.event.sender)}: {self.event.body}"
|
|
)
|
|
|
|
source_content = self.event.source["content"]
|
|
try:
|
|
alert_event_id = source_content["m.relates_to"]["m.in_reply_to"]["event_id"]
|
|
except KeyError:
|
|
logger.debug("Unable to find the event ID of the alert")
|
|
return
|
|
logger.debug(f"Read alert fingerprints for event {alert_event_id} from cache")
|
|
|
|
count_removed_silences = 0
|
|
alert_fingerprints = self.cache[alert_event_id]
|
|
for alert_fingerprint in alert_fingerprints:
|
|
logger.debug(
|
|
f"Delete silence for alert with fingerprint {alert_fingerprint}"
|
|
)
|
|
try:
|
|
removed_silences = await self.alertmanager.delete_silences(
|
|
alert_fingerprint
|
|
)
|
|
count_removed_silences += len(removed_silences)
|
|
except AlertmanagerError as e:
|
|
logger.exception(f"Unable to delete silence: {e}", exc_info=e)
|
|
|
|
await send_text_to_room(
|
|
self.client,
|
|
self.room.room_id,
|
|
f"Removed {count_removed_silences} silences.",
|
|
)
|
|
|
|
async def _react(self) -> None:
|
|
"""Make the bot react to the command message"""
|
|
# React with a start emoji
|
|
reaction = "⭐"
|
|
logger.debug(
|
|
f"Reacting with {reaction} to room {self.room.display_name} | "
|
|
f"{self.room.user_name(self.event.sender)}: {self.event.body}"
|
|
)
|
|
await react_to_event(
|
|
self.client, self.room.room_id, self.event.event_id, reaction
|
|
)
|
|
|
|
# React with some generic text
|
|
reaction = "Some text"
|
|
await react_to_event(
|
|
self.client, self.room.room_id, self.event.event_id, reaction
|
|
)
|
|
|
|
async def _show_help(self) -> None:
|
|
"""Show the help text"""
|
|
logger.debug(
|
|
f"Displaying help to room {self.room.display_name} | "
|
|
f"{self.room.user_name(self.event.sender)}: {self.event.body}"
|
|
)
|
|
if not self.args:
|
|
text = (
|
|
"Hello, I am a bot made with matrix-nio! Use `help commands` to view "
|
|
"available commands."
|
|
)
|
|
await send_text_to_room(self.client, self.room.room_id, text)
|
|
return
|
|
|
|
topic = self.args[0]
|
|
if topic == "rules":
|
|
text = "These are the rules!"
|
|
elif topic == "commands":
|
|
text = "Available commands: ..."
|
|
else:
|
|
text = "Unknown help topic!"
|
|
await send_text_to_room(self.client, self.room.room_id, text)
|
|
|
|
async def _unknown_command(self) -> None:
|
|
logger.debug(
|
|
f"Sending unknown command response to room {self.room.display_name} | "
|
|
f"{self.room.user_name(self.event.sender)}: {self.event.body}"
|
|
)
|
|
await send_text_to_room(
|
|
self.client,
|
|
self.room.room_id,
|
|
f"Unknown command '{self.command}'. Try the 'help' command for more information.",
|
|
)
|