[pull] main from keephq:main

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}]
 

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]]:

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", "")

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"}]

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

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()