paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-02 13:03:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

alertmanager: Strengthen types using WildValue.

Browse Source 
Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg
2021-12-16 22:03:22 -08:00
committed by Tim Abbott
parent ac9a5ba894
commit d436f8098e
1 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
zerver/webhooks/alertmanager/view.py
View File

@@ -1,11 +1,12 @@
# Webhooks for external integrations.
from typing import Any, Dict, List
from typing import Dict, List
from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@@ -15,7 +16,7 @@ from zerver.models import UserProfile
def api_alertmanager_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: Dict[str, Any] = REQ(argument_type="body"),
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
name_field: str = REQ("name", default="instance"),
desc_field: str = REQ("desc", default="alertname"),
) -> HttpResponse:
@@ -25,15 +26,17 @@ def api_alertmanager_webhook(
labels = alert.get("labels", {})
annotations = alert.get("annotations", {})
name = labels.get(name_field, annotations.get(name_field, "(unknown)"))
desc = labels.get(desc_field, annotations.get(desc_field, f"<missing field: {desc_field}>"))
name = labels.get(name_field, annotations.get(name_field, "(unknown)")).tame(check_string)
desc = labels.get(
desc_field, annotations.get(desc_field, f"<missing field: {desc_field}>")
).tame(check_string)
url = alert.get("generatorURL").replace("tab=1", "tab=0")
url = alert["generatorURL"].tame(check_string).replace("tab=1", "tab=0")
body = f"{desc} ([graph]({url}))"
if name not in topics:
topics[name] = {"firing": [], "resolved": []}
topics[name][alert["status"]].append(body)
topics[name][alert["status"].tame(check_string)].append(body)
for topic, statuses in topics.items():
for status, messages in statuses.items():