[pull] main from keephq:main

docs/snippets/providers/grafana-snippet-autogenerated.mdx

@@ -1,4 +1,4 @@
-{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py 
+{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py
 Do not edit it manually, as it will be overwritten */}
 
 ## Authentication
@@ -16,20 +16,46 @@ Certain scopes may be required to perform specific actions or queries via the pr
 
 ## In workflows
 
-This provider can't be used as a "step" or "action" in workflows. If you want to use it, please let us know by creating an issue in the [GitHub repository](https://github.com/keephq/keep/issues).
+This provider can be used in workflows.
 
 
+As "step" to query data, example:
+```yaml
+steps:
+    - name: Query grafana
+      provider: grafana
+      config: "{{ provider.my_provider_name }}"
+      with:
+        datasource_uid: {value}  # uid of the datasource to query (required).
+        query: {value}  # full query object for datasource specific fields, merged last so it overrides the arguments below.
+        expr: {value}  # query expression for Prometheus-style datasources.
+        raw_sql: {value}  # SQL statement for SQL datasources.
+        start: {value}  # start of the Grafana time range, absolute or relative such as now-1h.
+        end: {value}  # end of the Grafana time range, absolute or relative such as now.
+        instant: {value}  # run a Prometheus instant query instead of a range query.
+        max_data_points: {value}  # cap on the number of returned points.
+```
+
+
+
+
+
+Check the following workflow examples:
+- [create-new-incident-grafana-incident.yaml](https://github.com/keephq/keep/blob/main/examples/workflows/create-new-incident-grafana-incident.yaml)
+- [create_service_now_ticket_upon_alerts.yml](https://github.com/keephq/keep/blob/main/examples/workflows/create_service_now_ticket_upon_alerts.yml)
+- [query_grafana_loki.yaml](https://github.com/keephq/keep/blob/main/examples/workflows/query_grafana_loki.yaml)
+- [update-incident-grafana-incident.yaml](https://github.com/keephq/keep/blob/main/examples/workflows/update-incident-grafana-incident.yaml)
 
 
 ## Topology
-This provider pulls [topology](/overview/servicetopology) to Keep. It could be used in [correlations](/overview/correlation-topology) 
-and [mapping](/overview/enrichment/mapping#mapping-with-topology-data), and as a context 
+This provider pulls [topology](/overview/servicetopology) to Keep. It could be used in [correlations](/overview/correlation-topology)
+and [mapping](/overview/enrichment/mapping#mapping-with-topology-data), and as a context
 for [alerts](/alerts/sidebar#7-alert-topology-view) and [incidents](/overview#17-incident-topology).
 ## Connecting via Webhook (omnidirectional)
 This provider supports webhooks.
 
 If your Grafana is unreachable from Keep, you can use the following webhook url to configure Grafana to send alerts to Keep:
-    
+
     1. In Grafana, go to the Alerting tab in the Grafana dashboard.
     2. Click on Contact points in the left sidebar and create a new one.
     3. Give it a name and select Webhook as kind of contact point with webhook url as KEEP_BACKEND_URL/alerts/event/grafana.

keep/providers/grafana_provider/grafana_provider.py

@@ -23,6 +23,7 @@
     BaseTopologyProvider,
     ProviderHealthMixin,
 )
+from keep.exceptions.provider_exception import ProviderException
 from keep.providers.base.provider_exceptions import GetAlertException
 from keep.providers.grafana_provider.grafana_alert_format_description import (
     GrafanaAlertFormatDescription,
@@ -67,6 +68,7 @@ class GrafanaProviderAuthConfig:
 
 class GrafanaProvider(BaseTopologyProvider, ProviderHealthMixin):
     PROVIDER_DISPLAY_NAME = "Grafana"
+    QUERY_TIMEOUT = 60
     """Pull/Push alerts & Topology map from Grafana."""
 
     PROVIDER_CATEGORY = ["Monitoring", "Developer Tools"]
@@ -189,6 +191,96 @@ def get_provider_metadata(self) -> dict:
             "version": version,
         }
 
+    @staticmethod
+    def _frames_to_rows(frames: list[dict]) -> list[dict]:
+        """Flatten Grafana data frames into plain rows.
+
+        A frame is columnar - `schema.fields` names the columns and
+        `data.values` holds one array per column - which is awkward to
+        consume from a workflow, so turn it into a list of dicts.
+        """
+        rows = []
+        for frame in frames or []:
+            fields = frame.get("schema", {}).get("fields", [])
+            columns = frame.get("data", {}).get("values", [])
+            if not fields or not columns:
+                continue
+            names = [field.get("name") for field in fields]
+            for index in range(len(columns[0])):
+                rows.append(
+                    {
+                        name: columns[column_index][index]
+                        for column_index, name in enumerate(names)
+                        if column_index < len(columns)
+                    }
+                )
+        return rows
+
+    def _query(
+        self,
+        datasource_uid: str = "",
+        query: dict | None = None,
+        expr: str = "",
+        raw_sql: str = "",
+        start: str = "now-1h",
+        end: str = "now",
+        instant: bool = False,
+        max_data_points: int | None = None,
+        **kwargs: dict,
+    ) -> list[dict]:
+        """Query any datasource configured in Grafana via /api/ds/query.
+
+        This makes every datasource Grafana already knows about reachable
+        from a workflow with a single service account token, including ones
+        that have no dedicated Keep provider.
+
+        Args:
+            datasource_uid: uid of the datasource to query (required).
+            query: full query object for datasource specific fields, merged last so it overrides the arguments below.
+            expr: query expression for Prometheus-style datasources.
+            raw_sql: SQL statement for SQL datasources.
+            start: start of the Grafana time range, absolute or relative such as now-1h.
+            end: end of the Grafana time range, absolute or relative such as now.
+            instant: run a Prometheus instant query instead of a range query.
+            max_data_points: cap on the number of returned points.
+
+        Returns:
+            The result frames flattened into a list of row dicts.
+        """
+        if not datasource_uid:
+            raise ProviderException("datasource_uid is required")
+
+        target: dict = {"refId": "A", "datasource": {"uid": datasource_uid}}
+        if expr:
+            target["expr"] = expr
+        if raw_sql:
+            target["rawSql"] = raw_sql
+        if instant:
+            target["instant"] = True
+        if max_data_points:
+            target["maxDataPoints"] = max_data_points
+        if query:
+            target.update(query)
+
+        response = requests.post(
+            f"{self.authentication_config.host}/api/ds/query",
+            headers={"Authorization": f"Bearer {self.authentication_config.token}"},
+            json={"queries": [target], "from": start, "to": end},
+            timeout=self.QUERY_TIMEOUT,
+        )
+        if not response.ok:
+            raise ProviderException(
+                f"Failed to query datasource {datasource_uid}: "
+                f"{response.status_code} {response.text}"
+            )
+
+        result = response.json().get("results", {}).get(target["refId"], {})
+        if result.get("error"):
+            raise ProviderException(
+                f"Datasource {datasource_uid} returned an error: {result['error']}"
+            )
+        return self._frames_to_rows(result.get("frames", []))
+
     def get_alerts_configuration(self, alert_id: str | None = None):
         api = f"{self.authentication_config.host}/api/v1/provisioning/alert-rules"
         headers = {"Authorization": f"Bearer {self.authentication_config.token}"}
@@ -249,7 +341,7 @@ def calculate_fingerprint(alert: dict) -> str:
         if fingerprint:
             logger.debug("Fingerprint provided in alert")
             return fingerprint
-        
+
         labels = alert.get("labels", {})
         fingerprint = labels.get("fingerprint", "")
         if fingerprint:

keep/providers/litellm_provider/litellm_provider.py

@@ -60,6 +60,19 @@ def _format_messages(self, prompt: str) -> List[Dict[str, str]]:
         """Format the prompt as a chat message."""
         return [{"role": "user", "content": prompt}]
 
+    @staticmethod
+    def _strip_code_fence(text: str) -> str:
+        """Unwrap a ```json ... ``` fence, which models add unprompted."""
+        stripped = text.strip()
+        if not stripped.startswith("```"):
+            return text
+        stripped = stripped[3:]
+        if stripped.lower().startswith("json"):
+            stripped = stripped[4:]
+        if stripped.endswith("```"):
+            stripped = stripped[:-3]
+        return stripped.strip()
+
     def _query(
         self,
         prompt: str,
@@ -105,10 +118,22 @@ def _query(
             except KeyError:
                 generated_text = ""
 
+            # Reasoning models return content=None when the whole max_tokens
+            # budget went into reasoning tokens. Without this the step would
+            # silently succeed with {"response": None}.
+            if generated_text is None:
+                finish_reason = result["choices"][0].get("finish_reason")
+                raise ProviderException(
+                    "LiteLLM API returned no content "
+                    f"(finish_reason={finish_reason!r}). Reasoning models can "
+                    "spend the whole max_tokens budget on reasoning tokens; "
+                    "raise max_tokens or use a model that does not reason."
+                )
+
             # Try to parse as JSON if it's meant to be structured
             if structured_output_format:
                 try:
-                    generated_text = json.loads(generated_text)
+                    generated_text = json.loads(self._strip_code_fence(generated_text))
                 except json.JSONDecodeError:
                     raise ProviderException(
                         f"Failed to parse generated text as JSON: {generated_text}. Model not following the structured output format. Response: {result}"

tests/providers/grafana_provider/test_grafana_datasource_query.py

@@ -0,0 +1,105 @@
+"""Tests for querying Grafana datasources through /api/ds/query."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from keep.contextmanager.contextmanager import ContextManager
+from keep.exceptions.provider_exception import ProviderException
+from keep.providers.grafana_provider.grafana_provider import GrafanaProvider
+from keep.providers.models.provider_config import ProviderConfig
+
+PROM_FRAME = {
+    "schema": {"fields": [{"name": "Time"}, {"name": "Value"}]},
+    "data": {"values": [[1785959838794], [1611]]},
+}
+SQL_FRAME = {
+    "schema": {"fields": [{"name": "provider_id"}, {"name": "c"}]},
+    "data": {"values": [["a", "b"], [3884296, 1849312]]},
+}
+
+
+def _build_provider() -> GrafanaProvider:
+    config = ProviderConfig(
+        description="Grafana Provider",
+        authentication={"host": "https://grafana.example.com", "token": "t"},
+    )
+    return GrafanaProvider(ContextManager(tenant_id="test"), "grafana-test", config)
+
+
+def _response(payload, ok=True, status_code=200):
+    response = MagicMock()
+    response.ok = ok
+    response.status_code = status_code
+    response.text = "error body"
+    response.json = MagicMock(return_value=payload)
+    return response
+
+
+class TestFramesToRows:
+    def test_sql_frame_becomes_rows(self):
+        assert GrafanaProvider._frames_to_rows([SQL_FRAME]) == [
+            {"provider_id": "a", "c": 3884296},
+            {"provider_id": "b", "c": 1849312},
+        ]
+
+    def test_multiple_frames_are_concatenated(self):
+        rows = GrafanaProvider._frames_to_rows([PROM_FRAME, SQL_FRAME])
+        assert len(rows) == 3
+        assert rows[0] == {"Time": 1785959838794, "Value": 1611}
+
+    @pytest.mark.parametrize("frames", [[], [{}], [{"schema": {"fields": []}}]])
+    def test_empty_input_is_not_an_error(self, frames):
+        assert GrafanaProvider._frames_to_rows(frames) == []
+
+
+def test_query_builds_payload_and_returns_rows():
+    provider = _build_provider()
+    payload = {"results": {"A": {"frames": [SQL_FRAME]}}}
+    with patch("requests.post", return_value=_response(payload)) as post:
+        rows = provider._query(
+            datasource_uid="ds-uid", raw_sql="SELECT 1", start="now-6h", end="now"
+        )
+
+    assert rows[0]["provider_id"] == "a"
+    sent = post.call_args.kwargs["json"]
+    assert sent["from"] == "now-6h"
+    assert sent["queries"][0]["datasource"] == {"uid": "ds-uid"}
+    assert sent["queries"][0]["rawSql"] == "SELECT 1"
+
+
+def test_explicit_query_dict_overrides_defaults():
+    provider = _build_provider()
+    with patch(
+        "requests.post", return_value=_response({"results": {"A": {"frames": []}}})
+    ) as post:
+        provider._query(
+            datasource_uid="ds-uid",
+            expr="up",
+            query={"expr": "count(up)", "hide": True},
+        )
+
+    target = post.call_args.kwargs["json"]["queries"][0]
+    assert target["expr"] == "count(up)"
+    assert target["hide"] is True
+
+
+def test_missing_datasource_uid_raises():
+    provider = _build_provider()
+    with pytest.raises(ProviderException, match="datasource_uid is required"):
+        provider._query(expr="up")
+
+
+def test_http_error_raises():
+    provider = _build_provider()
+    with patch("requests.post", return_value=_response({}, ok=False, status_code=403)):
+        with pytest.raises(ProviderException, match="403"):
+            provider._query(datasource_uid="ds-uid", expr="up")
+
+
+def test_datasource_error_raises():
+    provider = _build_provider()
+    payload = {"results": {"A": {"error": "table not found"}}}
+    with patch("requests.post", return_value=_response(payload)):
+        with pytest.raises(ProviderException, match="table not found"):
+            provider._query(datasource_uid="ds-uid", raw_sql="SELECT 1")