Source code for spicerack.alerting

"""Alerting module."""
import logging
from contextlib import contextmanager
from datetime import timedelta
from typing import Iterator

from spicerack.administrative import Reason
from spicerack.alertmanager import AlertmanagerHosts
from spicerack.icinga import IcingaHosts

logger = logging.getLogger(__name__)

[docs]class AlertingHosts: """Operate on Alertmanager and Icinga via their APIs.""" def __init__( self, alertmanager_hosts: AlertmanagerHosts, icinga_hosts: IcingaHosts, ) -> None: """Initialize the instance. Arguments: alertmanager_hosts (spicerack.alertmanager.AlertmanagerHosts): the Alertmanager hosts to talk to. icinga_hosts (spicerack.icinga.IcingaHosts): the Icinga hosts to talk to. """ self._am = alertmanager_hosts self._icinga = icinga_hosts
[docs] @contextmanager def downtimed( self, reason: Reason, *, duration: timedelta = timedelta(hours=4), remove_on_error: bool = False ) -> Iterator[None]: """Context manager to perform actions while the hosts are downtimed on Alertmanager and Icinga. Arguments: reason (spicerack.administrative.Reason): the reason to set for the downtime. duration (datetime.timedelta, optional): the length of the downtime period. remove_on_error: should the downtime be removed even if an exception was raised. Yields: None: it just yields control to the caller once Alertmanager and Icinga have received the downtime and deletes the downtime once getting back the control. """ downtime_id = self.downtime(reason, duration=duration) try: yield except BaseException: if remove_on_error: self.remove_downtime(downtime_id) raise else: self.remove_downtime(downtime_id)
[docs] def downtime(self, reason: Reason, *, duration: timedelta = timedelta(hours=4)) -> str: """Issue a new downtime. Arguments: reason (spicerack.administrative.Reason): the downtime reason. duration (datetime.timedelta): how long to downtime for. Returns: str: the downtime ID. Raises: AlertmanagerError: if none of the `alertmanager_urls` API returned a success. IcingaError: if there is a problem downtiming the hosts in Icinga. """ self._icinga.downtime(reason, duration=duration) return self._am.downtime(reason, duration=duration)
[docs] def remove_downtime(self, downtime_id: str) -> None: """Remove a downtime. Arguments: downtime_id (str): the downtime ID to remove. """ self._icinga.remove_downtime() return self._am.remove_downtime(downtime_id)