9628bdc51f
This PR add labels for webhooks.
1. Make webhook "labelable" with ability to filter by labels.
2. Add labels to the webhook payload. It contain new field webhook with
it's name, id and labels. Field integration and alert_group has a
corresponding label field as well. See example of a new payload below:
```
{
"event": {
"type": "escalation"
},
"user": null,
"alert_group": {
"id": "IRFN6ZD31N31B",
"integration_id": "CTWM7U4A2QG97",
"route_id": "RUE7U7Z46SKGY",
"alerts_count": 1,
"state": "firing",
"created_at": "2023-11-22T08:54:55.178243Z",
"resolved_at": null,
"acknowledged_at": null,
"title": "Incident",
"permalinks": {
"slack": null,
"telegram": null,
"web": "http://grafana:3000/a/grafana-oncall-app/alert-groups/IRFN6ZD31N31B"
},
"labels": {
"severity": "critical"
}
},
"alert_group_id": "IRFN6ZD31N31B",
"alert_payload": {
"message": "This alert was sent by user for demonstration purposes"
},
"integration": {
"id": "CTWM7U4A2QG97",
"type": "webhook",
"name": "hi - Webhook",
"team": null,
"labels": {
"hello": "world",
"severity": "critical"
}
},
"notified_users": [],
"users_to_be_notified": [],
"webhook": {
"id": "WHAXK4BTC7TAEQ",
"name": "test",
"labels": {
"hello": "kesha"
}
}
}
```
I feel that there is an opportunity to make code cleaner - remove all
label logic from serializers, views and utils to models or dedicated
LabelerService and introduce Labelable interface with something like
label_verbal, update_labels methods. However, I don't want to tie
webhook labels with a refactoring.
---------
Co-authored-by: Dominik <dominik.broj@grafana.com>
88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
import typing
|
|
|
|
from django.apps import apps # noqa: I251
|
|
from django.conf import settings
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from apps.alerts.models import AlertGroup, AlertReceiveChannel
|
|
from apps.labels.models import AssociatedLabel
|
|
from apps.user_management.models import Organization
|
|
|
|
|
|
LABEL_OUTDATED_TIMEOUT_MINUTES = 30
|
|
ASSOCIATED_MODEL_NAME = "AssociatedLabel"
|
|
|
|
|
|
class LabelUpdateParam(typing.TypedDict):
|
|
name: str
|
|
|
|
|
|
class LabelParams(typing.TypedDict):
|
|
id: str
|
|
name: str
|
|
|
|
|
|
class LabelData(typing.TypedDict):
|
|
key: LabelParams
|
|
value: LabelParams
|
|
|
|
|
|
class ValueData(typing.TypedDict):
|
|
value_name: str
|
|
key_name: str
|
|
|
|
|
|
class LabelKeyData(typing.TypedDict):
|
|
key: LabelParams
|
|
values: typing.List[LabelParams]
|
|
|
|
|
|
LabelsData = typing.List[LabelData]
|
|
LabelsKeysData = typing.List[LabelParams]
|
|
|
|
|
|
def get_associating_label_model(obj_model_name: str) -> typing.Type["AssociatedLabel"]:
|
|
associating_label_model_name = obj_model_name + ASSOCIATED_MODEL_NAME
|
|
label_model = apps.get_model("labels", associating_label_model_name)
|
|
return label_model
|
|
|
|
|
|
def is_labels_feature_enabled(organization: "Organization") -> bool:
|
|
return (
|
|
settings.FEATURE_LABELS_ENABLED_FOR_ALL
|
|
or organization.org_id in settings.FEATURE_LABELS_ENABLED_FOR_GRAFANA_ORGS # Grafana org ID, not OnCall org ID
|
|
)
|
|
|
|
|
|
def assign_labels(alert_group: "AlertGroup", alert_receive_channel: "AlertReceiveChannel") -> None:
|
|
from apps.labels.models import AlertGroupAssociatedLabel
|
|
|
|
if not is_labels_feature_enabled(alert_receive_channel.organization):
|
|
return
|
|
|
|
# inherit labels from the integration
|
|
alert_group_labels = [
|
|
AlertGroupAssociatedLabel(
|
|
alert_group=alert_group,
|
|
organization=alert_receive_channel.organization,
|
|
key_name=label.key.name,
|
|
value_name=label.value.name,
|
|
)
|
|
for label in alert_receive_channel.labels.filter(inheritable=True).select_related("key", "value")
|
|
]
|
|
AlertGroupAssociatedLabel.objects.bulk_create(alert_group_labels)
|
|
|
|
|
|
def get_label_verbal(labelable) -> typing.Dict[str, str]:
|
|
"""
|
|
label_verbal returns dict of labels' key and values names for the given object
|
|
"""
|
|
return {label.key.name: label.value.name for label in labelable.labels.all().select_related("key", "value")}
|
|
|
|
|
|
def get_alert_group_label_verbal(alert_group: "AlertGroup") -> typing.Dict[str, str]:
|
|
"""
|
|
get_alert_group_label_verbal returns dict of labels' key and values names for the given alert group.
|
|
It's different from get_label_verbal, because AlertGroupAssociated labels store key/value_name, not key/value_id
|
|
"""
|
|
return {label.key_name: label.value_name for label in alert_group.labels.all()}
|