import os import json import logging from alerta.plugins import PluginBase LOG = logging.getLogger() class EnhanceGrafana(PluginBase): """ Plugin for parsing alerts coming from Grafana """ def pre_receive(self, alert, **kwargs): # Parse Grafana-specific fields alert.attributes['grafanaStatus'] = alert.raw_data.get('status', '') def htmlify(link: str, desc: str) -> str: return f'<a href="{link}" target="_blank">{desc}</a>'; # User-specified "Panel ID" annotation panelURL = alert.raw_data.get('panelURL', '') if panelURL: alert.attributes['grafanaPanelUrl'] = panelURL alert.attributes['grafanaPanelHtml'] = htmlify(panelURL, "Grafana Panel") # User-specified "Dashboard UID" annotation dashboardURL = alert.raw_data.get('dashboardURL', '') if dashboardURL: alert.attributes['grafanaDashboardUrl'] = dashboardURL alert.attributes['grafanaDashboardHtml'] = htmlify(dashboardURL, "Grafana Dashboard") alertURL = alert.raw_data.get('generatorURL', '') if alertURL: # expose alert view URL, as user may not have edit rights # Convert from # http://host:3000/alerting/kujybCynk/edit # to # http://host:3000/alerting/grafana/kujybCynk/view alertURL = alertURL.replace("/alerting/", "/alerting/grafana/").replace("/edit", "/view") alert.attributes['grafanaAlertUrl'] = alertURL alert.attributes['grafanaAlertHtml'] = htmlify(alertURL, "Grafana Alert") silenceURL = alert.raw_data.get('silenceURL', '') if silenceURL: alert.attributes['grafanaSilenceUrl'] = silenceURL alert.attributes['grafanaSilenceHtml'] = htmlify(silenceURL, "Grafana Silence Alert") return alert def post_receive(self, alert, **kwargs): return def status_change(self, alert, status, text, **kwargs): return def take_action(self, alert, action, text, **kwargs): raise NotImplementedError