matrix-alertbot/matrix_alertbot/alertmanager.py

import datetime
from typing import Dict, List

import diskcache
import pytimeparse
import requests
from requests import RequestException

from matrix_alertbot.config import Config
from matrix_alertbot.errors import (
    AlertmanagerError,
    AlertNotFoundError,
    SilenceNotFoundError,
)


class AlertmanagerClient:
    def __init__(self, config: Config) -> None:
        url = config.alertmanager_url
        self.api_url = f"{url}/api/v2"
        self.cache = diskcache.Cache(config.cache_dir)

    def get_alerts(self) -> List[Dict]:
        try:
            response = requests.get(f"{self.api_url}/alerts")
            response.raise_for_status()
        except RequestException as e:
            raise AlertmanagerError(f"Cannot fetch alerts from Alertmanager") from e
        return response.json()

    def get_alert(self, fingerprint: str) -> Dict:
        alerts = self.get_alerts()
        return self._find_alert(fingerprint, alerts)

    def get_alert_labels(self, fingerprint: str) -> Dict[str, str]:
        alert = self.get_alert(fingerprint)
        return alert["labels"]

    def create_silence(self, fingerprint: str, duration: str, user: str) -> str:
        labels = self.get_alert_labels(fingerprint)
        matchers = []
        for label_name, label_value in labels.items():
            matchers.append(
                {"name": label_name, "value": label_value, "isRegex": False}
            )

        start_time = datetime.datetime.now()
        duration_seconds = pytimeparse.parse(duration)
        duration_delta = datetime.timedelta(seconds=duration_seconds)
        end_time = start_time + duration_delta

        silence = {
            "matchers": matchers,
            "startsAt": start_time.isoformat(),
            "endsAt": end_time.isoformat(),
            "createdBy": user,
            "comment": "Acknowledge alert from Matrix",
        }
        try:
            response = requests.post(f"{self.api_url}/silences", json=silence)
            response.raise_for_status()
        except RequestException as e:
            raise AlertmanagerError(
                f"Cannot create silence for alert fingerprint {fingerprint}"
            ) from e
        data = response.json()
        return data["silenceID"]

    def delete_silence(self, fingerprint: str) -> None:
        alert = self.get_alert(fingerprint)

        alert_state = alert["status"]["state"]
        if alert_state != "suppressed":
            raise SilenceNotFoundError(
                f"Cannot find silences for alert fingerprint {fingerprint} in state {alert_state}"
            )

        silences = alert["status"]["silencedBy"]
        for silence in silences:
            try:
                response = requests.delete(f"{self.api_url}/silence/{silence}")
                response.raise_for_status()
            except RequestException as e:
                raise AlertmanagerError(
                    f"Cannot delete silence with ID {silence}"
                ) from e

    @staticmethod
    def _find_alert(fingerprint: str, alerts: List[Dict]) -> Dict:
        for alert in alerts:
            if alert["fingerprint"] == fingerprint:
                return alert
        raise AlertNotFoundError(f"Cannot find alert with fingerprint {fingerprint}")