Improve /webhook/grafana api formatting

app.py

 import requests
-from flask import Flask, request, make_response
-from config import url
+from flask import Flask, request, make_response, render_template
+
+from config import url, hookshot_params, flask_params, msg_templates, msg_template_default
+from validate import grafana_validate_incoming
+
+
+app = Flask(__name__, **flask_params)
 
-app = Flask(__name__)
 
 @app.route("/webhook/slack/<hook>", methods=['POST'])
 def slack(hook):
@@ -65,35 +69,43 @@ def slack(hook):
 
 @app.route("/webhook/grafana/<hook>", methods=['POST', 'PUT'])
 def grafana(hook):
-    plain = ''
-    html = ''
+    """
+    see https://grafana.com/docs/grafana/latest/alerting/alerting-rules/manage-contact-points/integrations/webhook-notifier/
+    todo:
+    - handle query params to select different templates
+    - handle empty -> default
+    """
+
+    args = request.args
+    if "template" in args.keys():
+        template_type = args.get("template")
+        if template_type not in msg_templates:
+            template_type = msg_template_default
+    else:
+        template_type = msg_template_default
+
+    if "version" in args.keys():
+        version = args.get("version")
+    else:
+        version = "v2"
+
+    print(args)
+    print(f"template: {template_type}, version: {version}")
+
     incoming = request.json
     print('Got incoming /grafana hook: ' + str(incoming))
 
-    title = str(incoming.get('title', ''))
-    rule_url = str(incoming.get('ruleUrl', ''))
-    rule_name = str(incoming.get('ruleName', ''))
-    message = str(incoming.get('message', ''))
-    state = str(incoming.get('state', ''))
-    eval_matches = incoming.get('evalMatches', [])
-
-    if title and rule_url and rule_name:
-        plain += title + ' ' + rule_url + ': ' + rule_name + ' (' + state + ')\n'
-        html += '<b><a href="' + rule_url + '">' + title + '</a></b>: ' + rule_name + ' (' + state + ')<br/>\n'
+    if not isinstance(incoming, dict):
+        incoming = dict()
 
-    if message:
-        plain += message + '\n'
-        html += message + '<br/>\n'
+    grafana_validate_incoming(incoming)
 
-    for eval_match in eval_matches:
-        metric = str(eval_match.get('metric', ''))
-        value = str(eval_match.get('value', ''))
-        if metric and value:
-            plain += metric + ': ' + value + '\n'
-            html += '<b>' + metric + '</b>: ' + value + '<br/>\n'
+    t = lambda fmt: f"{template_type}.{fmt}.jinja"
+    plain = render_template(t("txt"), **incoming)
+    html = render_template(t("html"), **incoming)
 
     if plain and html:
-        json = {'text':plain,'html':html}
+        json = {'text':plain,'html':html, **hookshot_params}
         print('Sending hookshot: ' + str(json))
         r = requests.post(url + hook, json=json)
     else:
@@ -102,5 +114,6 @@ def grafana(hook):
 
     return {"ok":True}
 
+
 if __name__ == "__main__":
-    app.run(port=9080, debug=True)
+    app.run(host="0.0.0.0", port=9080, debug=True)

config.py

 url = 'https://hookshot.mbot.ethz.ch/webhook/'
+flask_params = {"template_folder": "template"}
+hookshot_params = {"version": "v2", "msgtype": "m.notice"}
+
+msg_templates = ("default", "concise", "ultra_concise")
+msg_template_default = "ultra_concise"

template/concise.html.jinja

+{% if alerts | length %}
+{% for alert in alerts %}
+<b><a href="{{ alert.generatorURL }}">[{{ alert.status }}] {{ alert.labels.alertname }}</a>:</b>
+{% if alert.labels %}
+{% for key, value in alert.labels.items() %}
+{{ key }}={{ value }} {{ ", " if not loop.last else " " }}
+{% endfor %}
+{% endif %}
+(<a href="{{ alert.silenceURL }}">Silence</a>)<br>
+{% if alert.valuesForJinja %}
+{% for key, value in alert.valuesForJinja.items() %}
+<b>{{ key }}:</b> {{ value }}<br>
+{% endfor %}
+{% endif %}
+{% endfor %}
+{% endif %}

template/concise.txt.jinja

+{% if alerts | length %}{% for alert in alerts %}
+[{{ alert.status }}] {{ alert.labels.alertname }}:
+Alert: {{ alert.generatorURL }}
+Silence: {{ alert.silenceURL }}
+{% if alert.labels %}{% for key, value in alert.labels.items() %}{{ key }}={{ value }}, {% endfor %}{% endif %}
+{% if alert.valuesForJinja %}{% for key, value in alert.valuesForJinja.items() %}{{ key }}: {{ value }}, {% endfor %}{% endif %}
+___
+{% endfor %}{% endif %}

template/default.html.jinja

+<b>{{ status | upper }}:{{ noFiring }} ({{ commonLabels.values() | join(" ") }})</b>
+<p><a href="{{ externalUrl }}?orgId={{ orgId }}">Grafana Instance</a></p>
+{% if alerts | length %}{% for alert in alerts %}
+<p><b>{{ alert.labels.alertname }}</b> (Status: {{ alert.status }}):
+    <a href="{{ alert.generatorURL }}">View</a>, <a href="{{ alert.silenceURL }}">Silence</a>
+</p>
+
+{% if alert.valuesForJinja %}
+<b>Values</b>
+<table>
+    {% for key, value in alert.valuesForJinja.items() %}
+    <tr>
+        <td>{{ key }}</td>
+        <td>{{ value }}</td>
+    </tr>
+    {% endfor %}
+</table>
+{% endif %}
+
+{% if alert.labels %}
+<b>Labels</b>
+<table>
+    {% for key, value in alert.labels.items() %}
+    <tr>
+        <td>{{ key }}</td>
+        <td>{{ value }}</td>
+    </tr>
+    {% endfor %}
+</table>
+{% endif %}
+<p><i>(StartsAt: {{alert.startsAtParsed }}{% if alert.endsAtParsed %}, EndsAt: {{alert.endsAtParsed }}{% endif %})</i>
+</p>
+<br>
+{% endfor %}{% endif %}
\ No newline at end of file

template/default.txt.jinja

+{{ status | upper }}:{{ noFiring }} ({{ commonLabels.values() | join(" ") }})
+Grafana: {{ externalUrl }}?orgId={{ orgId }}
+
+{% if alerts | length %}
+{% for alert in alerts %}
+---
+Name: {{ alert.labels.alertname }}
+Status: {{ alert.status }}
+View alert: {{ alert.generatorURL }}
+Silence alert: {{ alert.silenceURL }}
+
+{% if alert.Values %}
+Values:
+{% for key, value in alert.Values.items() %}
+{{ key }}: {{ value }}
+{% endfor %}
+{% endif %}
+
+{% if alert.labels %}
+Labels
+{% for key, value in alert.labels.items() %}
+{{ key }}: {{ value }}
+{% endfor %}
+{% endif %}
+{% endfor %}
+{% endif %}

template/slack.html.jinja

+{% if color %}
+<font color="{{ color}}">
+{% endif %}
+
+{% if title and title_link %}
+<b><a href="{{ title_link }}">{{ title}}</a></b><br/>
+{% elif title %}
+<b>{{ title}}</b><br/>
+{% endif %}
+
+{% if text %}
+{{ text }}<br />
+{% endif %}
+
+{% for field in fields %}
+<b>{{ field.title }}:</b> {{ value}}<br/>
+{% endfor %}
+
+{% if footer %}
+{{ footer}}<br/>
+{% endif %}
+
+{% if color%}
+</font>
+{% endif %}
\ No newline at end of file

template/slack.txt.jinja

+{% if title and title_link %}
+{{ title }} {{ title_link }}
+{% elif title %}{{ title }}{% endif %}
+
+{% if text %}{{ text }}{% endif %}
+
+{% for field in fields %}
+{{ field.title }}: {{ value}}
+{% endfor %}
+
+{{ footer}}

template/ultra_concise.html.jinja

+{% if alerts | length %}
+{% for alert in alerts %}
+{{ "<br>" if not loop.first }}
+<b><a href="{{ alert.generatorURL }}">[{{ alert.status }}] {{ alert.labels.alertname }}</a></b> 
+(<a href="{{ alert.silenceURL }}">Silence</a>)<b>:</b>
+{% if alert.labels %}
+{% for key, value in alert.labels.items() %}
+{{ key }}={{ value }} {{- ", " if not loop.last else " " }}
+{% endfor %}
+{% endif %}
+<br>
+{% if alert.valuesForJinja %}
+{% for key, value in alert.valuesForJinja.items() %}
+<b>{{ key }}:</b> {{ value }}{{ ", " if not loop.last else " " }}
+{% endfor %}
+{% endif %}
+{% endfor %}
+{% endif %}

template/ultra_concise.txt.jinja

+{% if alerts | length %}{% for alert in alerts %}
+[{{ alert.status }}] {{ alert.labels.alertname }}:
+Alert: {{ alert.generatorURL }}
+Silence: {{ alert.silenceURL }}
+{% if alert.labels %}{% for key, value in alert.labels.items() %}{{ key }}={{ value }}, {% endfor %}{% endif %}
+{% if alert.valuesForJinja %}{% for key, value in alert.valuesForJinja.items() %}{{ key }}: {{ value }}, {% endfor %}{% endif %}
+___
+{% endfor %}{% endif %}

validate.py

+from datetime import datetime
+from typing import Dict
+
+# https://grafana.com/docs/grafana-cloud/alerting-and-irm/alerting/configure-notifications/manage-contact-points/integrations/webhook-notifier/
+
+
+def validate_str_entry(d: Dict, k: str):
+    """Empty string if not exists"""
+    d[k] = d.get(k, "")
+
+
+def validate_number_entry(d: Dict, k: str):
+    """0 if not exists"""
+    d[k] = d.get(k, 0)
+
+
+def validate_dict_entry(d: Dict, k: str):
+    """Empty dict if not exists"""
+    entry = d.get(k, None)
+    if not isinstance(entry, dict):
+        entry = dict()
+    d[k] = entry
+
+
+def grafana_validate_incoming(incoming: Dict):
+    """
+    Makes sure the necessary keys are present in the incoming dict so it can be parsed by jinja
+    """
+    string_keys = ("receiver", "status", "externalUrl", "version", "groupKey")
+    for key in string_keys:
+        validate_str_entry(incoming, key)
+
+    validate_number_entry(incoming, "orgId")
+    validate_number_entry(incoming, "truncatedAlerts")
+
+    validate_dict_entry(incoming, "groupLabels")
+    validate_dict_entry(incoming, "commonLabels")
+    validate_dict_entry(incoming, "commonAnnotations")
+
+    incoming["alerts"] = alerts = incoming.get("alerts", [])
+
+    for alert in alerts:  # Handle non dict values?
+        grafana_validate_alert(alert)
+
+    l = lambda a: a.get("status", "") == "firing"
+    incoming["no_firing"] = len(tuple(filter(l, alerts)))
+
+
+def grafana_validate_alert(alert: Dict):
+
+    string_keys = (
+        "status",
+        "startsAt",
+        "endsAt",
+        "generatorURL",
+        "fingerprint",
+        "silenceURL",
+        "imageURL",
+    )
+    for key in string_keys:
+        validate_str_entry(alert, key)
+
+    validate_dict_entry(alert, "labels")
+    validate_dict_entry(alert, "annotations")
+    validate_dict_entry(alert, "values")
+
+    # fixes "values" being shadowed in jinja
+    alert["valuesForJinja"] = alert.get("values", None)
+
+    alert["startsAtParsed"] = format_dt(alert.get("startsAt", ""))
+    alert["endsAtParsed"] = format_dt(alert.get("endsAt", ""))
+
+
+def format_dt(dt_str: str) -> str:
+    """
+    Returns empty string if error or is year 1 e.g. zero value of golang time.Time
+    see https://pkg.go.dev/time#Time
+    """
+    try:
+        dt = datetime.fromisoformat(dt_str)
+        if dt.year < 2:
+            return ""
+        return dt.strftime("%Y-%m-%d %H:%M:%S %z")
+    except ValueError:
+        return ""