alerta_grafana.py

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