from datetime import timedelta from datetime import datetime import logging import urllib.request import urllib.parse import json log = logging.getLogger('grafana') class Silencer(): def __init__(self, url, apikey): self._url = url self._apikey = apikey def silence_alert(self): with urllib.request.urlopen( urllib.request.Request( urllib.parse.urljoin( self._url, 'api/alertmanager/grafana/api/v2/silences', ), headers={ 'Authorization': f'Bearer {self._apikey}', }, ), data=bytes(json.dumps({ "comment": "Openfoam run", "createdBy": "Python OpenFoam run", "endsAt": (datetime.utcnow() \ + timedelta(days=1)).isoformat(), "matchers": [ { "isEqual": True, "isRegex": False, "name": "alertname", "value": "CPU Usage" } ], "startsAt": datetime.utcnow().isoformat() }), 'utf-8')) as api_req: api_res = api_req.read() self._alertid = json.loads(api_res)['id'] log.info(f'Grafana alerts silenced (id: {self._alertid})') def unsilence_alert(self): with urllib.request.urlopen( urllib.request.Request( urllib.parse.urljoin( self._url, 'api/alertmanager/grafana/api/v2/silence/' f'{self._alertid}', ), headers={ 'Authorization': f'Bearer {self._apikey}', }, method='DELETE')): log.info(f'Grafana silence (id: {self._alertid}) deleted')