From e015fdab0cd66c7277c4f97b105cf58bcc66426c Mon Sep 17 00:00:00 2001 From: Artem Rootman <4586640+artemrootman@users.noreply.github.com> Date: Thu, 6 Aug 2026 09:32:57 +0300 Subject: [PATCH 1/2] feat(grafana): query any datasource via /api/ds/query (#6673) --- .../grafana-snippet-autogenerated.mdx | 36 +++++- .../grafana_provider/grafana_provider.py | 94 +++++++++++++++- .../test_grafana_datasource_query.py | 105 ++++++++++++++++++ 3 files changed, 229 insertions(+), 6 deletions(-) create mode 100644 tests/providers/grafana_provider/test_grafana_datasource_query.py diff --git a/docs/snippets/providers/grafana-snippet-autogenerated.mdx b/docs/snippets/providers/grafana-snippet-autogenerated.mdx index d74c069934..dc8a0bb669 100644 --- a/docs/snippets/providers/grafana-snippet-autogenerated.mdx +++ b/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. diff --git a/keep/providers/grafana_provider/grafana_provider.py b/keep/providers/grafana_provider/grafana_provider.py index 5704be6e0e..14bd9a9883 100644 --- a/keep/providers/grafana_provider/grafana_provider.py +++ b/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: diff --git a/tests/providers/grafana_provider/test_grafana_datasource_query.py b/tests/providers/grafana_provider/test_grafana_datasource_query.py new file mode 100644 index 0000000000..380c99d79b --- /dev/null +++ b/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") From ce302f61cada7fb9c50d0261e810bce1b7abc3c2 Mon Sep 17 00:00:00 2001 From: Artem Rootman <4586640+artemrootman@users.noreply.github.com> Date: Thu, 6 Aug 2026 09:33:51 +0300 Subject: [PATCH 2/2] fix(litellm): fail loudly on null content and accept fenced JSON (#6671) --- .../litellm_provider/litellm_provider.py | 27 +++++- .../test_litellm_response_parsing.py | 90 +++++++++++++++++++ 2 files changed, 116 insertions(+), 1 deletion(-) create mode 100644 tests/providers/litellm_provider/test_litellm_response_parsing.py diff --git a/keep/providers/litellm_provider/litellm_provider.py b/keep/providers/litellm_provider/litellm_provider.py index 3cd565122e..32b22a4fc5 100644 --- a/keep/providers/litellm_provider/litellm_provider.py +++ b/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}" diff --git a/tests/providers/litellm_provider/test_litellm_response_parsing.py b/tests/providers/litellm_provider/test_litellm_response_parsing.py new file mode 100644 index 0000000000..6dd2f5bec3 --- /dev/null +++ b/tests/providers/litellm_provider/test_litellm_response_parsing.py @@ -0,0 +1,90 @@ +""" +Tests for how the LiteLLM provider parses a chat completion response. + +Two failure modes this covers, both silent or misleading before the fix: + +1. A reasoning model can return `content: null` when the whole `max_tokens` + budget went into reasoning tokens. The provider returned + `{"response": None}` and the workflow step counted as successful. +2. Models frequently wrap structured output in a ```json fence, which made + `json.loads()` fail and killed the step even though the JSON was valid. +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.litellm_provider.litellm_provider import LitellmProvider +from keep.providers.models.provider_config import ProviderConfig + +SCHEMA = { + "type": "object", + "properties": {"verdict": {"type": "string"}}, + "required": ["verdict"], +} + + +def _build_provider() -> LitellmProvider: + config = ProviderConfig( + description="LiteLLM Provider", + authentication={"api_url": "https://litellm.example.com/v1"}, + ) + return LitellmProvider(ContextManager(tenant_id="test"), "litellm-test", config) + + +def _response(content, finish_reason="stop"): + response = MagicMock() + response.raise_for_status = MagicMock() + response.json = MagicMock( + return_value={ + "choices": [ + {"message": {"content": content}, "finish_reason": finish_reason} + ] + } + ) + return response + + +class TestStripCodeFence: + @pytest.mark.parametrize( + "text", + [ + '```json\n{"verdict": "ok"}\n```', + '```\n{"verdict": "ok"}\n```', + ' ```json\n{"verdict": "ok"}\n``` ', + ], + ) + def test_fenced_json_is_unwrapped(self, text): + assert LitellmProvider._strip_code_fence(text) == '{"verdict": "ok"}' + + def test_plain_json_is_untouched(self): + assert LitellmProvider._strip_code_fence('{"verdict": "ok"}') == ( + '{"verdict": "ok"}' + ) + + +def test_structured_output_accepts_fenced_json(): + provider = _build_provider() + with patch( + "requests.post", return_value=_response('```json\n{"verdict": "ok"}\n```') + ): + result = provider._query(prompt="hi", structured_output_format=SCHEMA) + assert result["response"] == {"verdict": "ok"} + + +def test_none_content_raises_instead_of_returning_null(): + provider = _build_provider() + with patch("requests.post", return_value=_response(None, finish_reason="length")): + with pytest.raises(ProviderException) as excinfo: + provider._query(prompt="hi") + message = str(excinfo.value) + assert "no content" in message + assert "length" in message + + +def test_plain_text_response_still_works(): + provider = _build_provider() + with patch("requests.post", return_value=_response("plain answer")): + assert provider._query(prompt="hi")["response"] == "plain answer"