| import os |
| |
| RESOLVED_STATUSES = ('UP', 'OK', 'resolved') |
| |
| |
| def create_file(file_path): |
| if not os.path.exists(file_path): |
| with open(file_path, 'w+'): |
| return file_path |
| |
| |
| def _format_subject(alert, add_links=False): |
| if add_links: |
| severity = alert['labels']['severity'].upper() |
| name = alert['labels']['alertname'] |
| return '[{}] {}'.format(severity, name) |
| |
| subject = alert['annotations']['summary'] |
| host = alert['labels'].get('host') |
| if host is not None: |
| return '[{}] {}'.format(host, subject) |
| return subject |
| |
| |
| def _remove_advice(description): |
| return description.split(' Modify the')[0] |
| |
| |
| def _get_prometheus_link(alert): |
| # switch from Console to Graph tab to show trends |
| condition_url = alert['generatorURL'].replace('g0.tab=1', 'g0.tab=0') |
| |
| # list alert firing instances in the cluster |
| # via query: ALERTS{alertname="<alertname>",alertstate="firing"} |
| alert_query = ('g1.range_input=1h&' |
| 'g1.expr=ALERTS%7Balertname%3D%22' |
| '{}' |
| '%22%2Calertstate%3D%22firing%22%7D&' |
| 'g1.tab=0').format(alert['labels']['alertname']) |
| return '{}&{}'.format(condition_url, alert_query) |
| |
| |
| def _format_description(alert, add_links=False): |
| description_old = _remove_advice(alert['annotations']['description']) |
| |
| if not add_links: |
| return description_old |
| |
| msg = ('To check trends of the alert underlying condition metric ' |
| ' and alert occurrence click on Prometheus UI link') |
| |
| prometheus_url = _get_prometheus_link(alert) |
| return '{}\n\n{}:\n{}'.format(description_old, msg, prometheus_url) |
| |
| |
| def alert_fields_and_action(alert, add_links=False, cluster_id=None): |
| fields = [] |
| |
| if cluster_id is not None: |
| if alert['labels'].get('cluster_id') is None: |
| alert['labels']['cluster_id'] = cluster_id |
| |
| if alert['status'] in RESOLVED_STATUSES: |
| action = 'close_case' |
| fields.append(alert['labels']) |
| else: |
| action = 'create_case' |
| # Order of keys matters |
| fields.append(_format_subject(alert, add_links)) |
| fields.append(_format_description(alert, add_links)) |
| fields.append(alert['labels']) |
| return fields, action |
| |
| |
| def is_true(name): |
| if isinstance(name, str): |
| return name.lower() == 'true' |
| return name is True |
| |
| def get_env_or_secret(env, secret): |
| # returns the value of the env var if present, |
| # otherwise docker swarm secret if present, |
| # null when neither of the two present. |
| value = os.getenv(env, 'null') |
| if value != 'null': |
| return value |
| fpath = f'/run/secrets/{secret}' |
| exist = os.path.exists(fpath) |
| if exist: |
| with open(fpath) as file: |
| value = file.read().rstrip('\n') |
| return value |