.gitignore

@@ -2,3 +2,4 @@
 .bash*
 .profile*
 env/
+.flaskenv
\ No newline at end of file

README.md

+# Webhook to Matrix Hookshot
+
+This API translates incoming webhooks into generic webhooks (hookshot).
+
+1. Receive Grafana webhook alert
+2. Prepare alert data (error handling, add additional values)
+3. Apply data to selected template (html, text)
+4. Send out notification to hookshot
+
+## Configuration
+
+`https://webhooks.mbot.ethz.ch/webhook/grafana/<hook>?template=<template>&version=<version>`
+
+| Param      | Description                     | Optional |
+| ---------- | ------------------------------- | -------- |
+| `hook`     | Hookshot ID of your webhook     |          |
+| `template` | Template name (see table below) | Yes      |
+| `version`  | API version (currently ignored) | Yes      |
+
+This is the URL needed by Grafana. Create a new contact point and paste it.
+
+## How to run
+
+`flask --app app --debug run`
+
+Set hostname and port with `-h` and `-p` respectively.
+
+## Templates
+
+| Name             | Description                 |
+| ---------------- | --------------------------- |
+| `default`        | shows important values      |
+| `oneliner`       | `default` but more concise  |
+| `detailed`       | `default` but better        |
+| `detailed_table` | `detailed` with html tables |
+
+For mor details see [templates](/templates.md).
+
+## Jinja Templates
+
+| `<template_name>` |                                                           |
+| ----------------- | --------------------------------------------------------- |
+| Path HTML         | `template/<template_name>.html.jinja`                     |
+| Path Text         | `template/<template_name>.txt.jinja`                      |
+| Registration      | Append `<template_name>` to `msg_templates` `(config.py)` |
+
+- Info
+  - Template rendering context (variables available in the template):
+    - Incoming grafana alert data
+    - Additional values (see below)
+  - Error handling: missing/invalid data is interpreted as empty/zero
+- Keep in mind
+  - Use `alert['values']` (and not `alert.values`) to keep Jinja happy
+  - Only use html tags allowed by the matrix spec
+  - Remove unnecessary whitespace
+
+### Rendering Context
+
+You can use every value provided by Grafana. For a complete list see [Grafana Docs](https://grafana.com/docs/grafana/latest/alerting/configure-notifications/manage-contact-points/integrations/webhook-notifier/).
+
+#### Reserved / Special Values
+
+These values come directly from Grafana but have a special meaning or are used to
+compute additional values and are thus explained here.
+
+|                     | Value                              | Description |
+| ------------------- | ---------------------------------- | ----------- |
+| Reserved Labels     | `alert.labels['alertname']`        |             |
+|                     | `alert.labels['grafana_folder']`   |             |
+| Special Annotations | `alert.annotations['description']` |             |
+|                     | `alert.annotations['summary]`      |             |
+| Timestamps          | `alert['startsAt']`                |             |
+|                     | `alert['endsAt']`                  |             |
+
+**Note:** Reserved labels / special annotations also apply to `commonLabels`, `commonAnnotations`
+
+#### Additional Values
+
+These values are added to the incoming Grafana alert which is then applied to the Jinja template
+
+```jsonc
+{
+  "normalLabels": {},
+  "normalAnnotations": {},
+  "alerts": [
+    {
+      "uniqueLabels": {},
+      "uniqueAnnotations": {},
+      "valuesForJinja": {},
+      "startsAtParsed": "",
+      "endsAtParsed": "",
+    },
+    (...)
+  ],
+  "commonStartsAtParsed": "",
+  "commonEndsAtParsed": "",
+}
+```
+
+| Value                  | Description                                      |
+| ---------------------- | ------------------------------------------------ |
+| `normalLabels`         | Labels that are not "reserved"                   |
+| `normalAnnotations`    | Annotations that are not "special"               |
+| `uniqueLabels`         | Labels not in `commonLabels`                     |
+| `uniqueAnnotations`    | Annotations not in `commonAnnotations`           |
+| `startsAtParsed`       | Formatted version of `startsAt`                  |
+| `endsAtParsed`         | Formatted version of `endsAt`                    |
+| `commonStartsAtParsed` | `startsAtParsed` if same on every alert instance |
+| `commonEndsAtParsed`   | `endsAtParsed` if same on every alert instance   |
+
+### Inherit Template
+
+Inherit from another text template if you only want to change html
+
+```jinja
+{%- include "detailed.txt.jinja" -%}
+```
+
+## Links
+
+- [Matrix message spec](https://spec.matrix.org/latest/client-server-api/#mroommessage-msgtypes)
+- [Hookshot webhook handling](https://matrix-org.github.io/matrix-hookshot/latest/setup/webhooks.html#webhook-handling)
+- Grafana
+  - [Alerts webhook format](https://grafana.com/docs/grafana/latest/alerting/configure-notifications/manage-contact-points/integrations/webhook-notifier/)
+  - [Labels and annotations](https://grafana.com/docs/grafana/next/alerting/fundamentals/alert-rules/annotation-label/)

app.py

 import requests
-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
+from flask import Flask, make_response, render_template, request
 
+from config import (
+    flask_params,
+    hookshot_params,
+    msg_template_default,
+    msg_templates,
+    url,
+)
+from sanitize import grafana_sanitize_incoming
 
 app = Flask(__name__, **flask_params)
+app.jinja_env.lstrip_blocks = True
+app.jinja_env.trim_blocks = True
 
 
-@app.route("/webhook/slack/<hook>", methods=['POST'])
+@app.route("/webhook/slack/<hook>", methods=["POST"])
 def slack(hook):
-    plain = ''
-    html = ''
+    plain = ""
+    html = ""
     incoming = request.json
-    print('Got incoming /slack hook: ' + str(incoming))
+    print("Got incoming /slack hook: " + str(incoming))
 
-    attachments = incoming.get('attachments', [])
-    username = str(incoming.get('username', ''))
+    attachments = incoming.get("attachments", [])
+    username = str(incoming.get("username", ""))
 
     for attachment in attachments:
-        color = str(attachment.get('color', '')).lower()
-        title = str(attachment.get('title', ''))
-        title_link = str(attachment.get('title_link', ''))
-        text = str(attachment.get('text', ''))
-        footer = str(attachment.get('footer', ''))
-        fields = attachment.get('fields', [])
+        color = str(attachment.get("color", "")).lower()
+        title = str(attachment.get("title", ""))
+        title_link = str(attachment.get("title_link", ""))
+        text = str(attachment.get("text", ""))
+        footer = str(attachment.get("footer", ""))
+        fields = attachment.get("fields", [])
 
-        html += '<font color="' + color + '">' if color else ''
+        html += '<font color="' + color + '">' if color else ""
 
         if title and title_link:
-            plain += title + ' ' + title_link + '\n'
-            html += '<b><a href="' + title_link + '">' + title + '</a></b><br/>\n'
+            plain += title + " " + title_link + "\n"
+            html += '<b><a href="' + title_link + '">' + title + "</a></b><br/>\n"
         elif title:
-            plain += title + '\n'
-            html += '<b>' + title + '</b><br/>\n'
+            plain += title + "\n"
+            html += "<b>" + title + "</b><br/>\n"
 
         if text:
-            plain += text + '\n'
-            html += text + '<br/>\n'
+            plain += text + "\n"
+            html += text + "<br/>\n"
 
         for field in fields:
-            title = str(field.get('title', ''))
-            value = str(field.get('value', ''))
+            title = str(field.get("title", ""))
+            value = str(field.get("value", ""))
             if title and value:
-                plain += title + ': ' + value + '\n'
-                html += '<b>' + title + '</b>: ' + value + '<br/>\n'
+                plain += title + ": " + value + "\n"
+                html += "<b>" + title + "</b>: " + value + "<br/>\n"
 
         if footer:
-            plain += footer + '\n'
-            html += footer + '<br/>\n'
+            plain += footer + "\n"
+            html += footer + "<br/>\n"
 
-        html += '</font>' if color else ''
+        html += "</font>" if color else ""
 
     if plain and html:
         if username:
-            json = {'text':plain,'html':html,'username':username}
+            json = {"text": plain, "html": html, "username": username}
         else:
-            json = {'text':plain,'html':html}
-        print('Sending hookshot: ' + str(json))
+            json = {"text": plain, "html": html}
+        print("Sending hookshot: " + str(json))
         r = requests.post(url + hook, json=json)
     else:
-        print('Invalid format, sending unmodified.')
+        print("Invalid format, sending unmodified.")
         r = requests.post(url + hook, json=incoming)
 
-    response = make_response('ok', 200)
+    response = make_response("ok", 200)
     response.mimetype = "text/plain"
     return response
 
-@app.route("/webhook/grafana/<hook>", methods=['POST', 'PUT'])
-def grafana(hook):
-    """
-    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
-    """
 
+@app.route("/webhook/grafana/<hook>", methods=["POST", "PUT"])
+def grafana(hook):
     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 = args.get("template", msg_template_default)
+    if template_type not in msg_templates:
         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}")
+    version = args.get("version", "v2")  # unused at the moment
 
     incoming = request.json
-    print('Got incoming /grafana hook: ' + str(incoming))
+    print("Got incoming /grafana hook: " + str(incoming))
 
-    if not isinstance(incoming, dict):
-        incoming = dict()
-
-    grafana_validate_incoming(incoming)
-
-    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, **hookshot_params}
-        print('Sending hookshot: ' + str(json))
-        r = requests.post(url + hook, json=json)
-    else:
-        print('Invalid format, sending incoming as str.')
-        r = requests.post(url + hook, json={'text':'Invalid format: ' + str(incoming)})
+    sanitized = grafana_sanitize_incoming(incoming)
 
-    return {"ok":True}
+    plain = render_template(f"{template_type}.txt.jinja", **sanitized)
+    html = render_template(f"{template_type}.html.jinja", **sanitized)
 
+    json = {"text": plain, "html": html, **hookshot_params}
+    print("Sending hookshot: " + str(json))
+    r = requests.post(url + hook, json=json)
 
-if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=9080, debug=True)
+    return {"ok": True}

config.py

-url = 'https://hookshot.mbot.ethz.ch/webhook/'
+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"
+
+reserved_label_keys = ["grafana_folder", "alertname"]
+special_annotation_keys = ["description", "summary"]
+
+reserved_labels = {k: "" for k in reserved_label_keys}
+special_annotations = {k: "" for k in special_annotation_keys}
+
+incoming_default = {
+    "receiver": "",
+    "status": "",
+    "orgId": 0,
+    "alerts": [],
+    "groupLabels": {},
+    "commonLabels": {},
+    "commonAnnotatons": {},
+    "externalUrl": "",
+    "version": "",
+    "groupKey": "",
+    "commonStartsAtParsed": "",  # computed
+}
+
+alert_default = {
+    "status": "",
+    "labels": {},
+    "annotations": {},
+    "startsAt": "",
+    "endsAt": "",
+    "values": {},
+    "generatorURL": "",
+    "fingerprint": "",
+    "silenceURL": "",
+    "imageURL": "",
+    "uniqueLabels": "",  # computed
+    "uniqueAnnotations": "",  # computed
+    "startsAtParsed": "",  # computed
+    "endsAtParsed": "",  # computed
+}
+
+msg_templates = ("default", "oneliner", "detailed", "detailed_table")
+msg_template_default = "default"

sanitize.py

+from datetime import datetime
+from typing import Dict, List, Tuple
+
+from config import alert_default, incoming_default, reserved_labels, special_annotations
+
+
+def get_unique_dict(source: Dict, common: Tuple[str]) -> Dict:
+    """Filters out entries present in common keys"""
+    return {k: v for k, v in source.items() if k not in common}
+
+
+def get_common_time(alerts: List[Dict], key: str) -> str | None:
+    """Return formatted timestamp if is common across alert instances"""
+    unique_starts_at = {a[key] for a in alerts}
+    if len(unique_starts_at) == 1:
+        return unique_starts_at.pop()
+
+
+def merge_with_default_dict(source_raw: Dict, default: Dict) -> Dict:
+    """Merges source dict with default dict. Handles source_raw being None"""
+    source = source_raw if isinstance(source_raw, dict) else dict()
+    return default | source
+
+
+def grafana_sanitize_incoming(incoming: Dict) -> Dict:
+    sanitized = merge_with_default_dict(incoming, incoming_default)
+
+    sanitized["commonLabels"] = merge_with_default_dict(
+        sanitized["commonLabels"], reserved_labels
+    )
+
+    sanitized["commonAnnotations"] = merge_with_default_dict(
+        sanitized["commonAnnotations"], special_annotations
+    )
+
+    sanitized["alerts"] = [
+        grafana_sanitize_alert(
+            alert,
+            sanitized["commonLabels"].keys(),
+            sanitized["commonAnnotations"].keys(),
+        )
+        for alert in incoming.get("alerts", [])
+    ]
+
+    firing_alerts = [a for a in sanitized["alerts"] if a["status"] == "firing"]
+    sanitized["numberOfFiring"] = len(firing_alerts)
+
+    for special_key in special_annotations.keys():
+        sanitized[special_key] = sanitized["commonAnnotations"].get(special_key, "")
+
+    sanitized["normalAnnotations"] = {
+        k: v
+        for k, v in sanitized["commonAnnotations"].items()
+        if k not in special_annotations
+    }
+    sanitized["normalLabels"] = {
+        k: v for k, v in sanitized["commonLabels"].items() if k not in reserved_labels
+    }
+
+    sanitized["commonStartsAtParsed"] = get_common_time(
+        sanitized["alerts"], "startsAtParsed"
+    )
+    sanitized["commonEndsAtParsed"] = get_common_time(
+        sanitized["alerts"], "endsAtParsed"
+    )
+
+    return sanitized
+
+
+def grafana_sanitize_alert(
+    alert: Dict, common_label_keys: Tuple[str], common_annotation_keys: Tuple[str]
+) -> Dict:
+    sanitized = alert_default | alert
+
+    sanitized["values"] = merge_with_default_dict(sanitized["values"], dict())
+
+    sanitized["uniqueLabels"] = get_unique_dict(sanitized["labels"], common_label_keys)
+    sanitized["uniqueAnnotations"] = get_unique_dict(
+        sanitized["annotations"], common_annotation_keys
+    )
+
+    sanitized["startsAtParsed"] = format_dt(sanitized.get("startsAt", ""))
+    sanitized["endsAtParsed"] = format_dt(sanitized.get("endsAt", ""))
+
+    return sanitized
+
+
+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 ""

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
+{%- if alerts | length %}{% for alert in alerts -%}
+<b><a href="{{ alert['generatorURL'] }}">[{{ alert['status'] }}] {{ alert['labels']['alertname'] }}</a></b>
+{%- for key, value in alert['uniqueLabels'].items() -%}
+{%- if key != 'alertname' %}
+{{- " " + key }}={{ value }} {{ "," if not loop.last }}
+{%- endif %}
+{%- endfor -%}
+&#32;(<a href="{{ alert['silenceURL'] }}">Silence</a>)<br>
+{%- if alert['values'] %}
+{%- for key, value in alert['values'].items() -%}
+<b>{{ key }}:</b> {{ value }}<br>
+{%- endfor %}
+{%- endif %}
+{%- endfor %}{% endif %}
\ No newline at end of file