Webhooks trigger tasks on alert group events (#1533)

engine/apps/webhooks/apps.py

@@ -3,3 +3,6 @@ from django.apps import AppConfig
 
 class WebhooksConfig(AppConfig):
     name = "apps.webhooks"
+
+    def ready(self):
+        from . import signals  # noqa: F401

engine/apps/webhooks/listeners.py

@@ -0,0 +1,24 @@
+import logging
+
+from django.apps import apps
+
+from .tasks import alert_group_created, alert_group_status_change
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+def on_alert_created(**kwargs):
+    Alert = apps.get_model("alerts", "Alert")
+    alert_pk = kwargs["alert"]
+    alert = Alert.objects.get(pk=alert_pk)
+
+    alert_group_created.apply_async((alert.group_id,))
+
+
+def on_action_triggered(**kwargs):
+    AlertGroupLogRecord = apps.get_model("alerts", "AlertGroupLogRecord")
+    log_record = kwargs["log_record"]
+    if not isinstance(log_record, AlertGroupLogRecord):
+        log_record = AlertGroupLogRecord.objects.get(pk=log_record)
+    alert_group_status_change.apply_async((log_record.type, log_record.alert_group_id, log_record.author_id))

engine/apps/webhooks/signals.py

@@ -0,0 +1,6 @@
+from apps.alerts.signals import alert_create_signal, alert_group_action_triggered_signal
+
+from .listeners import on_action_triggered, on_alert_created
+
+alert_create_signal.connect(on_alert_created)
+alert_group_action_triggered_signal.connect(on_action_triggered)

engine/apps/webhooks/tasks/__init__.py

@@ -0,0 +1,2 @@
+from .alert_group_status import alert_group_created, alert_group_status_change  # noqa: F401
+from .trigger_webhook import execute_webhook, send_webhook_event  # noqa: F401

engine/apps/webhooks/tasks/alert_group_status.py

@@ -0,0 +1,91 @@
+import logging
+
+from celery.utils.log import get_task_logger
+from django.conf import settings
+
+from apps.alerts.models import AlertGroup, AlertGroupLogRecord
+from apps.user_management.models import User
+from apps.webhooks.models import Webhook
+from apps.webhooks.utils import serialize_event
+from common.custom_celery_tasks import shared_dedicated_queue_retry_task
+
+from .trigger_webhook import send_webhook_event
+
+logger = get_task_logger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+MAX_RETRIES = 10
+
+
+@shared_dedicated_queue_retry_task(
+    bind=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=1 if settings.DEBUG else MAX_RETRIES
+)
+def alert_group_created(self, alert_group_id):
+    try:
+        alert_group = AlertGroup.unarchived_objects.get(pk=alert_group_id)
+    except AlertGroup.DoesNotExist:
+        return
+
+    trigger_type = Webhook.TRIGGER_NEW
+    event = {
+        "type": "Firing",
+        "time": alert_group.started_at,
+    }
+    data = serialize_event(event, alert_group, None)
+    organization_id = alert_group.channel.organization_id
+    team_id = alert_group.channel.team_id
+    send_webhook_event.apply_async(
+        (trigger_type, data), kwargs={"organization_id": organization_id, "team_id": team_id}
+    )
+
+
+@shared_dedicated_queue_retry_task(
+    bind=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=1 if settings.DEBUG else MAX_RETRIES
+)
+def alert_group_status_change(self, action_type, alert_group_id, user_id):
+    try:
+        alert_group = AlertGroup.unarchived_objects.get(pk=alert_group_id)
+        user = User.objects.filter(pk=user_id).first()
+    except (AlertGroup.DoesNotExist):
+        return
+
+    # TODO: update mapping, maybe use a dict instead
+    if action_type == AlertGroupLogRecord.TYPE_ACK:
+        trigger_type = Webhook.TRIGGER_ACKNOWLEDGE
+        event = {
+            "type": "Acknowledge",
+            "time": alert_group.acknowledged_at,
+        }
+    elif action_type == AlertGroupLogRecord.TYPE_RESOLVED:
+        trigger_type = Webhook.TRIGGER_RESOLVE
+        event = {
+            "type": "Resolve",
+            "time": alert_group.resolved_at,
+        }
+    elif action_type == AlertGroupLogRecord.TYPE_SILENCE:
+        trigger_type = Webhook.TRIGGER_SILENCE
+        event = {
+            "type": "Silence",
+            "time": alert_group.silenced_at,
+            "until": alert_group.silenced_until,
+        }
+    elif action_type == AlertGroupLogRecord.TYPE_UN_SILENCE:
+        trigger_type = Webhook.TRIGGER_UNSILENCE
+        event = {
+            "type": "Unsilence",
+        }
+    elif action_type == AlertGroupLogRecord.TYPE_UN_RESOLVED:
+        trigger_type = Webhook.TRIGGER_UNRESOLVE
+        event = {
+            "type": "Unresolve",
+        }
+    else:
+        return
+
+    data = serialize_event(event, alert_group, user)
+    organization_id = alert_group.channel.organization_id
+    team_id = alert_group.channel.team_id
+    send_webhook_event.apply_async(
+        (trigger_type, data), kwargs={"organization_id": organization_id, "team_id": team_id}
+    )

engine/apps/webhooks/tasks/trigger_webhook.py

@@ -0,0 +1,88 @@
+import json
+import logging
+from json import JSONDecodeError
+
+from celery.utils.log import get_task_logger
+from django.apps import apps
+from django.conf import settings
+from django.utils import timezone
+
+from apps.webhooks.models import WebhookLog
+from apps.webhooks.utils import InvalidWebhookData, InvalidWebhookHeaders, InvalidWebhookTrigger, InvalidWebhookUrl
+from common.custom_celery_tasks import shared_dedicated_queue_retry_task
+
+logger = get_task_logger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+@shared_dedicated_queue_retry_task(
+    autoretry_for=(Exception,), retry_backoff=True, max_retries=1 if settings.DEBUG else None
+)
+def send_webhook_event(trigger_type, data, team_id=None, organization_id=None):
+    Webhooks = apps.get_model("webhooks", "Webhook")
+    webhooks_qs = Webhooks.objects.filter(trigger_type=trigger_type, organization_id=organization_id, team_id=team_id)
+
+    for webhook in webhooks_qs:
+        execute_webhook.apply_async((webhook.pk, data))
+
+
+@shared_dedicated_queue_retry_task(
+    autoretry_for=(Exception,), retry_backoff=True, max_retries=1 if settings.DEBUG else None
+)
+def execute_webhook(webhook_pk, data):
+    Webhooks = apps.get_model("webhooks", "Webhook")
+
+    status = {
+        "last_run_at": timezone.now(),
+        "input_data": data,
+        "url": None,
+        "trigger": None,
+        "headers": None,
+        "data": None,
+        "response_status": None,
+        "response": None,
+    }
+
+    exception = None
+    try:
+        webhook = Webhooks.objects.get(pk=webhook_pk)
+        triggered, status["trigger"] = webhook.check_trigger(data)
+        if triggered:
+            status["url"] = webhook.build_url(data)
+            request_kwargs = webhook.build_request_kwargs(data, raise_data_errors=True)
+            status["headers"] = json.dumps(request_kwargs.get("headers", {}))
+            if webhook.forward_all:
+                status["data"] = "All input_data forwarded as payload"
+            elif "json" in request_kwargs:
+                status["data"] = json.dumps(request_kwargs["json"])
+            else:
+                status["data"] = request_kwargs.get("data")
+            response = webhook.make_request(status["url"], request_kwargs)
+            status["response_status"] = response.status_code
+            try:
+                status["response"] = json.dumps(response.json())
+            except JSONDecodeError:
+                status["response"] = response.content.decode("utf-8")
+        else:
+            # do not add a log entry if the webhook is not triggered
+            return
+    except Webhooks.DoesNotExist:
+        logger.warn(f"Webhook {webhook_pk} does not exist")
+        return
+    except InvalidWebhookUrl as e:
+        status["url"] = e.message
+    except InvalidWebhookTrigger as e:
+        status["trigger"] = e.message
+    except InvalidWebhookHeaders as e:
+        status["headers"] = e.message
+    except InvalidWebhookData as e:
+        status["data"] = e.message
+    except Exception as e:
+        status["response"] = str(e)
+        exception = e
+
+    # create/update log entry
+    WebhookLog.objects.update_or_create(webhook_id=webhook_pk, defaults=status)
+
+    if exception:
+        raise exception

engine/apps/webhooks/tests/test_alert_group_status_change.py

@@ -0,0 +1,154 @@
+from unittest.mock import call, patch
+
+import pytest
+from django.utils import timezone
+
+from apps.alerts.models import AlertGroup, AlertGroupLogRecord
+from apps.public_api.serializers import IncidentSerializer
+from apps.webhooks.models import Webhook
+from apps.webhooks.tasks import alert_group_created, alert_group_status_change
+
+
+@pytest.mark.django_db
+def test_alert_group_created(make_organization, make_alert_receive_channel, make_alert_group):
+    organization = make_organization()
+    alert_receive_channel = make_alert_receive_channel(organization)
+    alert_group = make_alert_group(alert_receive_channel)
+
+    with patch("apps.webhooks.tasks.trigger_webhook.send_webhook_event.apply_async") as mock_send_event:
+        alert_group_created(alert_group.pk)
+
+    assert mock_send_event.called
+    expected_data = {
+        "event": {
+            "type": "Firing",
+            "time": alert_group.started_at,
+        },
+        "user": None,
+        "alert_group": IncidentSerializer(alert_group).data,
+        "alert_group_id": alert_group.public_primary_key,
+        "alert_payload": "",
+    }
+
+    assert mock_send_event.call_args == call(
+        (Webhook.TRIGGER_NEW, expected_data), kwargs={"organization_id": organization.pk, "team_id": None}
+    )
+
+
+@pytest.mark.django_db
+def test_alert_group_created_for_team(make_organization, make_team, make_alert_receive_channel, make_alert_group):
+    organization = make_organization()
+    team = make_team(organization)
+    alert_receive_channel = make_alert_receive_channel(organization, team=team)
+    alert_group = make_alert_group(alert_receive_channel)
+
+    with patch("apps.webhooks.tasks.trigger_webhook.send_webhook_event.apply_async") as mock_send_event:
+        alert_group_created(alert_group.pk)
+
+    assert mock_send_event.called
+    expected_data = {
+        "event": {
+            "type": "Firing",
+            "time": alert_group.started_at,
+        },
+        "user": None,
+        "alert_group": IncidentSerializer(alert_group).data,
+        "alert_group_id": alert_group.public_primary_key,
+        "alert_payload": "",
+    }
+
+    assert mock_send_event.call_args == call(
+        (Webhook.TRIGGER_NEW, expected_data), kwargs={"organization_id": organization.pk, "team_id": team.pk}
+    )
+
+
+@pytest.mark.django_db
+def test_alert_group_created_does_not_exist():
+    assert AlertGroup.all_objects.filter(pk=53).first() is None
+
+    with patch("apps.webhooks.tasks.trigger_webhook.send_webhook_event.apply_async") as mock_send_event:
+        alert_group_created(53)
+
+    assert not mock_send_event.called
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "action_type,event_type,webhook_type,time_field",
+    [
+        (AlertGroupLogRecord.TYPE_ACK, "Acknowledge", Webhook.TRIGGER_ACKNOWLEDGE, "acknowledged_at"),
+        (AlertGroupLogRecord.TYPE_RESOLVED, "Resolve", Webhook.TRIGGER_RESOLVE, "resolved_at"),
+        (AlertGroupLogRecord.TYPE_SILENCE, "Silence", Webhook.TRIGGER_SILENCE, "silenced_at"),
+        (AlertGroupLogRecord.TYPE_UN_SILENCE, "Unsilence", Webhook.TRIGGER_UNSILENCE, None),
+        (AlertGroupLogRecord.TYPE_UN_RESOLVED, "Unresolve", Webhook.TRIGGER_UNRESOLVE, None),
+    ],
+)
+def test_alert_group_status_change(
+    make_organization,
+    make_user_for_organization,
+    make_alert_receive_channel,
+    make_alert_group,
+    action_type,
+    event_type,
+    webhook_type,
+    time_field,
+):
+    organization = make_organization()
+    user = make_user_for_organization(organization)
+    alert_receive_channel = make_alert_receive_channel(organization)
+    alert_group = make_alert_group(alert_receive_channel)
+
+    with patch("apps.webhooks.tasks.trigger_webhook.send_webhook_event.apply_async") as mock_send_event:
+        alert_group_status_change(action_type, alert_group.pk, user.pk)
+
+    expected_data = {
+        "event": {
+            "type": event_type,
+        },
+        "user": user.username,
+        "alert_group": IncidentSerializer(alert_group).data,
+        "alert_group_id": alert_group.public_primary_key,
+        "alert_payload": "",
+    }
+    if time_field is not None:
+        expected_data["event"]["time"] = getattr(alert_group, time_field)
+    if action_type == AlertGroupLogRecord.TYPE_SILENCE:
+        expected_data["event"]["until"] = alert_group.silenced_until
+    assert mock_send_event.call_args == call(
+        (webhook_type, expected_data), kwargs={"organization_id": organization.pk, "team_id": None}
+    )
+
+
+@pytest.mark.django_db
+def test_alert_group_status_change_does_not_exist():
+    assert AlertGroup.all_objects.filter(pk=53).first() is None
+
+    with patch("apps.webhooks.tasks.trigger_webhook.send_webhook_event.apply_async") as mock_send_event:
+        alert_group_status_change(AlertGroupLogRecord.TYPE_ACK, 53, None)
+
+    assert not mock_send_event.called
+
+
+@pytest.mark.django_db
+def test_alert_group_status_change_for_team(make_organization, make_team, make_alert_receive_channel, make_alert_group):
+    organization = make_organization()
+    team = make_team(organization)
+    alert_receive_channel = make_alert_receive_channel(organization, team=team)
+    alert_group = make_alert_group(alert_receive_channel, resolved=True, resolved_at=timezone.now())
+
+    with patch("apps.webhooks.tasks.trigger_webhook.send_webhook_event.apply_async") as mock_send_event:
+        alert_group_status_change(AlertGroupLogRecord.TYPE_RESOLVED, alert_group.pk, None)
+
+    expected_data = {
+        "event": {
+            "type": "Resolve",
+            "time": alert_group.resolved_at,
+        },
+        "user": None,
+        "alert_group": IncidentSerializer(alert_group).data,
+        "alert_group_id": alert_group.public_primary_key,
+        "alert_payload": "",
+    }
+    assert mock_send_event.call_args == call(
+        (Webhook.TRIGGER_RESOLVE, expected_data), kwargs={"organization_id": organization.pk, "team_id": team.pk}
+    )

engine/apps/webhooks/tests/test_trigger_webhook.py

@@ -0,0 +1,177 @@
+import json
+from unittest.mock import call, patch
+
+import pytest
+
+from apps.webhooks.models import Webhook
+from apps.webhooks.tasks import execute_webhook, send_webhook_event
+
+
+class MockResponse:
+    def __init__(self, status_code=200):
+        self.status_code = status_code
+
+    def json(self):
+        return {"response": self.status_code}
+
+
+@pytest.mark.django_db
+def test_send_webhook_event_filters(make_organization, make_team, make_custom_webhook):
+    organization = make_organization()
+    other_organization = make_organization()
+    other_team = make_team(organization)
+
+    webhooks = {}
+    for trigger_type, _ in Webhook.TRIGGER_TYPES:
+        webhooks[trigger_type] = make_custom_webhook(
+            organization=organization,
+            trigger_type=trigger_type,
+        )
+
+    other_team_webhook = make_custom_webhook(
+        organization=organization, team=other_team, trigger_type=Webhook.TRIGGER_ACKNOWLEDGE
+    )
+    other_org_webhook = make_custom_webhook(organization=other_organization, trigger_type=Webhook.TRIGGER_NEW)
+
+    sample_data = {"field": "value"}
+    for trigger_type, _ in Webhook.TRIGGER_TYPES:
+        with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
+            send_webhook_event(trigger_type, sample_data, organization_id=organization.pk)
+        assert mock_execute.call_args == call((webhooks[trigger_type].pk, sample_data))
+
+    # other team
+    with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
+        send_webhook_event(
+            Webhook.TRIGGER_ACKNOWLEDGE, sample_data, organization_id=organization.pk, team_id=other_team.pk
+        )
+    assert mock_execute.call_args == call((other_team_webhook.pk, sample_data))
+
+    # other org
+    with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
+        send_webhook_event(Webhook.TRIGGER_NEW, sample_data, organization_id=other_organization.pk)
+    assert mock_execute.call_args == call((other_org_webhook.pk, sample_data))
+
+
+@pytest.mark.django_db
+def test_execute_webhook_ok(make_organization, make_custom_webhook):
+    # set trigger, build_url, build_requests_args, check status/log
+    organization = make_organization()
+    webhook = make_custom_webhook(
+        organization=organization,
+        url="https://something/{{ alert_id }}/",
+        http_method="POST",
+        trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
+        trigger_template="{{ integration_id == 'the-integration' }}",
+        headers='{"some-header": "{{ alert_id }}"}',
+        data='{"value": "{{ value }}"}',
+        forward_all=False,
+    )
+    data = {
+        "integration_id": "the-integration",
+        "alert_id": "ID123",
+        "value": "42",
+    }
+
+    mock_response = MockResponse()
+    with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
+        mock_gethostbyname.return_value = "8.8.8.8"
+        with patch("apps.webhooks.models.webhook.requests") as mock_requests:
+            mock_requests.post.return_value = mock_response
+            execute_webhook(webhook.pk, data)
+
+    assert mock_requests.post.called
+    expected_call = call(
+        "https://something/ID123/",
+        timeout=10,
+        headers={"some-header": "ID123"},
+        json={"value": "42"},
+    )
+    assert mock_requests.post.call_args == expected_call
+    # check logs
+    log = webhook.logs.all()[0]
+    assert log.response_status == "200"
+    assert log.response == json.dumps(mock_response.json())
+    assert log.input_data == data
+    assert log.data == json.dumps({"value": "42"})
+    assert log.headers == json.dumps({"some-header": "ID123"})
+    assert log.url == "https://something/ID123/"
+
+
+@pytest.mark.django_db
+def test_execute_webhook_trigger_false(make_organization, make_custom_webhook):
+    # set trigger, build_url, build_requests_args, check status/log
+    organization = make_organization()
+    webhook = make_custom_webhook(
+        organization=organization,
+        url="https://something/{{ alert_id }}/",
+        http_method="POST",
+        trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
+        trigger_template="{{ integration_id == 'the-integration' }}",
+    )
+    data = {
+        "integration_id": "other-integration",
+        "alert_id": "ID123",
+        "value": "42",
+    }
+
+    with patch("apps.webhooks.models.webhook.requests") as mock_requests:
+        execute_webhook(webhook.pk, data)
+
+    assert not mock_requests.post.called
+    # check no logs
+    assert webhook.logs.count() == 0
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "field_name,value,log_field_name,expected_error",
+    [
+        (
+            "url",
+            "https://myserver/{{ alert_payload.id }}/triggered",
+            "url",
+            "URL - Template Warning: 'alert_payload' is undefined",
+        ),
+        (
+            "trigger_template",
+            "{{ }}",
+            "trigger",
+            "Trigger - Template Error: Expected an expression, got 'end of print statement'",
+        ),
+        ("headers", '"{{foo|invalid}}"', "headers", "Headers - Template Error: No filter named 'invalid'."),
+        ("data", "{{ }}", "data", "Data - Template Error: Expected an expression, got 'end of print statement'"),
+    ],
+)
+def test_execute_webhook_errors(
+    make_organization, make_custom_webhook, field_name, value, log_field_name, expected_error
+):
+    organization = make_organization()
+    extra_kwargs = {field_name: value}
+    if "url" not in extra_kwargs:
+        extra_kwargs["url"] = "https://something.cool/"
+    webhook = make_custom_webhook(
+        organization=organization,
+        http_method="POST",
+        trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
+        forward_all=False,
+        **extra_kwargs,
+    )
+    data = {
+        "integration_id": "other-integration",
+        "alert_id": "ID123",
+        "value": "42",
+    }
+
+    with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
+        # make it a valid URL when resolving name
+        mock_gethostbyname.return_value = "8.8.8.8"
+        with patch("apps.webhooks.models.webhook.requests") as mock_requests:
+            execute_webhook(webhook.pk, data)
+
+    assert not mock_requests.post.called
+    log = webhook.logs.all()[0]
+    assert log.response_status is None
+    assert log.response is None
+    assert log.input_data == data
+    error = getattr(log, log_field_name)
+    assert error == expected_error

engine/apps/webhooks/utils.py

@@ -111,3 +111,21 @@ class EscapeDoubleQuotesDict(dict):
         if '"' in original_str:
             return re.sub('(?<!\\\\)"', '\\\\"', original_str)
         return original_str
+
+
+def serialize_event(event, alert_group, user):
+    from apps.public_api.serializers import IncidentSerializer
+
+    alert_payload = alert_group.alerts.first()
+    alert_payload_raw = ""
+    if alert_payload:
+        alert_payload_raw = alert_payload.raw_request_data
+
+    data = {
+        "event": event,
+        "user": user.username if user else None,
+        "alert_group": IncidentSerializer(alert_group).data,
+        "alert_group_id": alert_group.public_primary_key,
+        "alert_payload": alert_payload_raw,
+    }
+    return data