60 lines
2 KiB
Python
60 lines
2 KiB
Python
|
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')
|