kellya/matrix-webhook
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Support Grafana v9.x messages, fixes #29

Browse source
This commit is contained in:
Sven Seeberg 2022-09-07 15:01:45 +02:00
parent 3288c3ff5a
commit e139c3a61b
No known key found for this signature in database
GPG key ID: 29559DD5A83806B5
4 changed files with 157 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
matrix_webhook/formatters.py
View file

@ -4,8 +4,10 @@ import re
def grafana(data, headers):
"""Pretty-print a grafana notification."""
"""Pretty-print a Grafana (version 8 and older) notification."""
text = ""
if "ruleName" not in data and "alerts" in data:
return grafana_9x(data, headers)
if "title" in data:
text = "#### " + data["title"] + "\n"
if "message" in data:
@ -17,6 +19,17 @@ def grafana(data, headers):
return data
def grafana_9x(data, headers):
"""Pretty-print a Grafana newer than v9.x notification."""
text = ""
if "title" in data:
text = "#### " + data["title"] + "\n"
if "message" in data:
text = text + data["message"].replace("\n", "\n\n") + "\n\n"
data["body"] = text
return data
def github(data, headers):
"""Pretty-print a github notification."""
# TODO: Write nice useful formatters. This is only an example.

41
tests/example_grafana_9x.json Normal file
View file

@ -0,0 +1,41 @@
{
"receiver": "",
"status": "firing",
"alerts": [
{
"status": "firing",
"labels": {
"alertname": "TestAlert",
"instance": "Grafana"
},
"annotations": {
"summary": "Notification test"
},
"startsAt": "2022-09-07T15:00:26.722304913+02:00",
"endsAt": "0001-01-01T00:00:00Z",
"generatorURL": "",
"fingerprint": "57c6d9296de2ad39",
"silenceURL": "https://grafana.example.com/alerting/silence/new?alertmanager=grafana&matcher=alertname%3DTestAlert&matcher=instance%3DGrafana",
"dashboardURL": "",
"panelURL": "",
"valueString": "[ metric='foo' labels={instance=bar} value=10 ]"
}
],
"groupLabels": {},
"commonLabels": {
"alertname": "TestAlert",
"instance": "Grafana"
},
"commonAnnotations": {
"summary": "Notification test"
},
"externalURL": "https://grafana.example.com/",
"version": "1",
"groupKey": "{alertname=\"TestAlert\", instance=\"Grafana\"}2022-09-07 15:00:26.722304913 +0200 CEST m=+246580.963796811",
"truncatedAlerts": 0,
"orgId": 1,
"title": "[FIRING:1] (TestAlert Grafana)",
"state": "alerting",
"message": "**Firing**\n\nValue: [ metric='foo' labels={instance=bar} value=10 ]\nLabels:\n - alertname = TestAlert\n - instance = Grafana\nAnnotations:\n - summary = Notification test\nSilence: https://grafana.example.com/alerting/silence/new?alertmanager=grafana&matcher=alertname%3DTestAlert&matcher=instance%3DGrafana\n",
"key": "ak"
}

51
tests/test_grafana_9x.py Normal file
View file

@ -0,0 +1,51 @@
"""
Test module for grafana v9 formatter.
ref https://grafana.com/docs/grafana/latest/alerting/old-alerting/notifications/#webhook
"""
import unittest
import httpx
import nio
from .start import BOT_URL, FULL_ID, MATRIX_ID, MATRIX_PW, MATRIX_URL
class Grafana9xFormatterTest(unittest.IsolatedAsyncioTestCase):
"""Grafana formatter test class."""
async def test_grafana_body(self):
"""Send a markdown message, and check the result."""
messages = []
client = nio.AsyncClient(MATRIX_URL, MATRIX_ID)
await client.login(MATRIX_PW)
room = await client.room_create()
with open("tests/example_grafana_9x.json") as f:
example_grafana_request = f.read()
self.assertEqual(
httpx.post(
f"{BOT_URL}/{room.room_id}",
params={"formatter": "grafana_9x"},
content=example_grafana_request,
).json(),
{"status": 200, "ret": "OK"},
)
sync = await client.sync()
messages = await client.room_messages(room.room_id, sync.next_batch)
await client.close()
message = messages.chunk[0]
self.assertEqual(message.sender, FULL_ID)
expected_body = (
"#### [FIRING:1] (TestAlert Grafana)\n**Firing**\n\n\n\nValue: [ metr"
"ic='foo' labels={instance=bar} value=10 ]\n\nLabels:\n\n - alertname "
"= TestAlert\n\n - instance = Grafana\n\nAnnotations:\n\n - summary = "
"Notification test\n\nSilence: https://grafana.example.com/alerting/si"
"lence/new?alertmanager=grafana&matcher=alertname%3DTestAlert&matcher="
"instance%3DGrafana\n\n\n\n"
)
self.assertEqual(message.body, expected_body)

51
tests/test_grafana_forward.py Normal file
View file

@ -0,0 +1,51 @@
"""
Test version 9 compatibility of grafana formatter.
ref https://grafana.com/docs/grafana/latest/alerting/old-alerting/notifications/#webhook
"""
import unittest
import httpx
import nio
from .start import BOT_URL, FULL_ID, MATRIX_ID, MATRIX_PW, MATRIX_URL
class GrafanaForwardFormatterTest(unittest.IsolatedAsyncioTestCase):
"""Grafana formatter test class."""
async def test_grafana_body(self):
"""Send a markdown message, and check the result."""
messages = []
client = nio.AsyncClient(MATRIX_URL, MATRIX_ID)
await client.login(MATRIX_PW)
room = await client.room_create()
with open("tests/example_grafana_9x.json") as f:
example_grafana_request = f.read()
self.assertEqual(
httpx.post(
f"{BOT_URL}/{room.room_id}",
params={"formatter": "grafana"},
content=example_grafana_request,
).json(),
{"status": 200, "ret": "OK"},
)
sync = await client.sync()
messages = await client.room_messages(room.room_id, sync.next_batch)
await client.close()
message = messages.chunk[0]
self.assertEqual(message.sender, FULL_ID)
expected_body = (
"#### [FIRING:1] (TestAlert Grafana)\n**Firing**\n\n\n\nValue: [ metr"
"ic='foo' labels={instance=bar} value=10 ]\n\nLabels:\n\n - alertname "
"= TestAlert\n\n - instance = Grafana\n\nAnnotations:\n\n - summary = "
"Notification test\n\nSilence: https://grafana.example.com/alerting/si"
"lence/new?alertmanager=grafana&matcher=alertname%3DTestAlert&matcher="
"instance%3DGrafana\n\n\n\n"
)
self.assertEqual(message.body, expected_body)