diff --git a/keep/providers/anthropic_provider/anthropic_provider.py b/keep/providers/anthropic_provider/anthropic_provider.py index dfd32d6b8b..c8bb20c6fd 100644 --- a/keep/providers/anthropic_provider/anthropic_provider.py +++ b/keep/providers/anthropic_provider/anthropic_provider.py @@ -90,7 +90,7 @@ def _query( system_prompt (str): System prompt override for this call. structured_output_format (dict): The structured output format to use. """ - client = Anthropic(api_key=self.authentication_config.api_key) + client = Anthropic(api_key=self.authentication_config.api_key) messages = [{"role": "user", "content": prompt}] diff --git a/keep/providers/base/base_provider.py b/keep/providers/base/base_provider.py index 47c701156d..2fa0ad3284 100644 --- a/keep/providers/base/base_provider.py +++ b/keep/providers/base/base_provider.py @@ -570,6 +570,20 @@ def get_alerts(self) -> list[AlertDto]: for alert in alerts: alert.providerId = self.provider_id alert.providerType = self.provider_type + + # Apply custom deduplication rules to pulled alerts + # (mirrors the logic in format_alert() for webhook alerts) + custom_deduplication_rule = get_custom_deduplication_rule( + tenant_id=self.context_manager.tenant_id, + provider_id=self.provider_id, + provider_type=self.provider_type, + ) + if custom_deduplication_rule: + for alert in alerts: + alert.fingerprint = self.get_alert_fingerprint( + alert, custom_deduplication_rule.fingerprint_fields + ) + return alerts def get_alerts_by_fingerprint(self, tenant_id: str) -> dict[str, list[AlertDto]]: diff --git a/keep/providers/grafana_provider/grafana_provider.py b/keep/providers/grafana_provider/grafana_provider.py index c98a940e5c..5704be6e0e 100644 --- a/keep/providers/grafana_provider/grafana_provider.py +++ b/keep/providers/grafana_provider/grafana_provider.py @@ -9,6 +9,7 @@ import logging import re import time +import urllib.parse import pydantic import requests @@ -288,6 +289,16 @@ def calculate_fingerprint(alert: dict) -> str: fingerprint = hashlib.sha256(fingerprint_string.encode()).hexdigest() return fingerprint + @staticmethod + def _resolve_alert_url(url: str | None, external_url: str | None) -> str | None: + if not url or not isinstance(url, str): + return None + if url.startswith(("http://", "https://")): + return url + if not external_url or not isinstance(external_url, str): + return None + return urllib.parse.urljoin(external_url, url) + @staticmethod def _format_alert( event: dict, provider_instance: "BaseProvider" = None @@ -324,8 +335,13 @@ def _format_alert( if values: extra["values"] = values - url = alert.get("generatorURL", None) - image_url = alert.get("imageURL", None) + external_url = event.get("externalURL") + url = GrafanaProvider._resolve_alert_url( + alert.get("generatorURL"), external_url + ) + image_url = GrafanaProvider._resolve_alert_url( + alert.get("imageURL"), external_url + ) # Always set these as "" when absent so workflow templates can # reference them safely without triggering render_context safe=True errors. dashboard_url = alert.get("dashboardURL", "") diff --git a/pyproject.toml b/pyproject.toml index e4e70695be..41fbabed47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "keep" -version = "0.51.0" +version = "0.52.0" description = "Alerting. for developers, by developers." authors = ["Keep Alerting LTD"] packages = [{include = "keep"}] diff --git a/tests/providers/grafana_provider/test_grafana_v12_webhook.py b/tests/providers/grafana_provider/test_grafana_v12_webhook.py new file mode 100644 index 0000000000..c4c327f476 --- /dev/null +++ b/tests/providers/grafana_provider/test_grafana_v12_webhook.py @@ -0,0 +1,103 @@ +from keep.providers.grafana_provider.grafana_provider import GrafanaProvider + + +def _grafana12_test_payload(generator_url: str = "?orgId=1") -> dict: + return { + "receiver": "webhook", + "status": "firing", + "alerts": [ + { + "status": "firing", + "labels": { + "alertname": "TestAlert", + "grafana_folder": "Test Folder", + "instance": "Grafana", + }, + "annotations": { + "description": "Test Description", + "summary": "TEST", + }, + "startsAt": "2026-05-05T13:04:57.62281314Z", + "endsAt": "0001-01-01T00:00:00Z", + "generatorURL": generator_url, + "fingerprint": "c4edaaf9f9f839b0", + "silenceURL": "https://grafana.example.com/alerting/silence/new", + "dashboardURL": "https://grafana.example.com/d/dashboard_uid", + "panelURL": "https://grafana.example.com/d/dashboard_uid?viewPanel=1", + "values": {"B": 22, "C": 1}, + "valueString": "[ var='B' value=22 ]", + "orgId": 1, + } + ], + "groupLabels": {"alertname": "TestAlert"}, + "commonLabels": {"alertname": "TestAlert"}, + "commonAnnotations": {"description": "Test Description"}, + "externalURL": "https://grafana.example.com/", + "appVersion": "12.4.2", + "version": "1", + "groupKey": "webhook-c4edaaf9f9f839b0-1777986297", + "truncatedAlerts": 0, + "orgId": 1, + "title": "TestAlert", + "state": "alerting", + "message": "Firing", + } + + +class TestResolveAlertUrl: + def test_absolute_url_passes_through(self): + assert ( + GrafanaProvider._resolve_alert_url( + "https://grafana.example.com/d/foo", "https://ignored.example.com/" + ) + == "https://grafana.example.com/d/foo" + ) + + def test_query_only_url_joins_with_external_url(self): + # Grafana 12 test alerts emit this shape. + assert ( + GrafanaProvider._resolve_alert_url( + "?orgId=1", "https://grafana.example.com/" + ) + == "https://grafana.example.com/?orgId=1" + ) + + def test_relative_path_joins_with_external_url(self): + assert ( + GrafanaProvider._resolve_alert_url( + "alerting/grafana/abc/view", "https://grafana.example.com/" + ) + == "https://grafana.example.com/alerting/grafana/abc/view" + ) + + def test_relative_url_without_external_url_returns_none(self): + assert GrafanaProvider._resolve_alert_url("?orgId=1", None) is None + assert GrafanaProvider._resolve_alert_url("?orgId=1", "") is None + + def test_missing_or_empty_url_returns_none(self): + assert GrafanaProvider._resolve_alert_url(None, "https://x/") is None + assert GrafanaProvider._resolve_alert_url("", "https://x/") is None + + +class TestFormatAlertGrafana12: + def test_test_alert_payload_parses(self): + event = _grafana12_test_payload() + + alerts = GrafanaProvider._format_alert(event) + + assert len(alerts) == 1 + alert = alerts[0] + assert alert.name == "TestAlert" + assert alert.fingerprint == "c4edaaf9f9f839b0" + assert str(alert.url) == "https://grafana.example.com/?orgId=1" + assert alert.values == {"B": 22, "C": 1} + assert alert.description == "Test Description" + + def test_test_alert_without_external_url_drops_invalid_generator_url(self): + event = _grafana12_test_payload() + event.pop("externalURL") + + alerts = GrafanaProvider._format_alert(event) + + assert len(alerts) == 1 + assert alerts[0].url is None diff --git a/tests/test_get_alerts_custom_dedup.py b/tests/test_get_alerts_custom_dedup.py new file mode 100644 index 0000000000..7809fa0adf --- /dev/null +++ b/tests/test_get_alerts_custom_dedup.py @@ -0,0 +1,141 @@ +import hashlib +import unittest +from unittest.mock import MagicMock, patch + +from keep.api.models.alert import AlertDto, AlertSeverity, AlertStatus +from keep.providers.base.base_provider import BaseProvider + + +def _make_alert(name, labels=None, fingerprint=None): + return AlertDto( + id=name, + name=name, + status=AlertStatus.FIRING, + severity=AlertSeverity.WARNING, + lastReceived="2025-01-01T00:00:00Z", + labels=labels or {}, + fingerprint=fingerprint, + source=["test"], + ) + + +class _StubProvider(BaseProvider): + """Concrete stub so we can instantiate BaseProvider for testing.""" + + def dispose(self): + pass + + def validate_config(self): + pass + + +def _make_provider(alerts): + """Create a minimal provider instance that returns canned alerts.""" + provider = object.__new__(_StubProvider) + provider.provider_id = "test-provider-id" + provider.provider_type = "prometheus" + provider.context_manager = MagicMock() + provider.context_manager.tenant_id = "test-tenant" + provider.logger = MagicMock() + provider._get_alerts = MagicMock(return_value=alerts) + return provider + + +class TestGetAlertsCustomDedup(unittest.TestCase): + @patch("keep.providers.base.base_provider.get_custom_deduplication_rule") + def test_custom_dedup_rule_overwrites_fingerprint(self, mock_get_rule): + """Pulled alerts should get fingerprints recalculated when a custom dedup rule exists.""" + alert_a = _make_alert( + "HighCPU", + labels={"alertname": "HighCPU", "env": "prod"}, + fingerprint="native-fp-1", + ) + alert_b = _make_alert( + "HighCPU", + labels={"alertname": "HighCPU", "env": "staging"}, + fingerprint="native-fp-1", # same native fingerprint + ) + + rule = MagicMock() + rule.fingerprint_fields = ["labels.alertname", "labels.env"] + mock_get_rule.return_value = rule + + provider = _make_provider([alert_a, alert_b]) + + with patch( + "keep.providers.base.base_provider.tracer" + ) as mock_tracer: + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock() + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock() + alerts = provider.get_alerts() + + # fingerprints should now differ because env differs + self.assertNotEqual(alerts[0].fingerprint, alerts[1].fingerprint) + + # verify fingerprint matches expected sha256 + expected_a = hashlib.sha256() + expected_a.update(b"HighCPU") + expected_a.update(b"prod") + self.assertEqual(alerts[0].fingerprint, expected_a.hexdigest()) + + expected_b = hashlib.sha256() + expected_b.update(b"HighCPU") + expected_b.update(b"staging") + self.assertEqual(alerts[1].fingerprint, expected_b.hexdigest()) + + @patch("keep.providers.base.base_provider.get_custom_deduplication_rule") + def test_no_custom_rule_keeps_original_fingerprint(self, mock_get_rule): + """Without a custom dedup rule, pulled alerts keep their original fingerprint.""" + alert = _make_alert( + "DiskFull", + labels={"alertname": "DiskFull"}, + fingerprint="original-fp", + ) + + mock_get_rule.return_value = None + + provider = _make_provider([alert]) + + with patch( + "keep.providers.base.base_provider.tracer" + ) as mock_tracer: + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock() + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock() + alerts = provider.get_alerts() + + self.assertEqual(alerts[0].fingerprint, "original-fp") + + @patch("keep.providers.base.base_provider.get_custom_deduplication_rule") + def test_custom_dedup_with_dot_notation_fields(self, mock_get_rule): + """Custom dedup should support dot-notation to access nested dict fields.""" + alert = _make_alert( + "NodeDown", + labels={"alertname": "NodeDown", "env_dc": "us-east", "group": "infra"}, + ) + + rule = MagicMock() + rule.fingerprint_fields = [ + "labels.alertname", + "labels.env_dc", + "labels.group", + ] + mock_get_rule.return_value = rule + + provider = _make_provider([alert]) + + with patch( + "keep.providers.base.base_provider.tracer" + ) as mock_tracer: + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock() + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock() + alerts = provider.get_alerts() + + expected = hashlib.sha256() + expected.update(b"NodeDown") + expected.update(b"us-east") + expected.update(b"infra") + self.assertEqual(alerts[0].fingerprint, expected.hexdigest()) + + +if __name__ == "__main__": + unittest.main()