feat(grafana): add deployment-annotations tool for change correlation (#2689)

Add a read-only GrafanaAnnotationsTool plus a query_annotations() method on
GrafanaClientBase so the agent can answer "did a deploy/config change precede this
alert?" for deploys from any source (ArgoCD/Flux, Helm, Terraform, manual), not just
GitHub pushes. Complements GitDeployTimelineTool and reuses the existing Grafana auth.

- service: query_annotations() mirrors query_alert_rules() (direct requests.get ->
  list[dict]); _map_annotation/_epoch_ms_to_iso map /api/annotations to ISO-8601 UTC
- tool: mirrors GrafanaAlertRulesTool, reuses GrafanaLogsTool helpers, supports the
  grafana_backend fixture path, forwards basic-auth credentials
- backends: add query_annotations() to the GrafanaBackend Protocol and all implementers
- docs: docs/grafana_annotations.mdx registered in docs.json
- tests: tests/tools/test_grafana_annotations_tool.py (schema, availability, extraction,
  backend path, UTC parsing, basic-auth, time-window override)

Author: kespineira

app/integrations/opensre/csv_grafana_backend.py

@@ -260,6 +260,10 @@ def query_traces(self, service_name: str = "", **_: Any) -> dict[str, Any]:
         traces = [{"traceID": tid, "spans": spans} for tid, spans in span_by_trace.items()]
         return {"traces": traces, "metrics": {}}
 
+    def query_annotations(self, **_: Any) -> list[dict[str, Any]]:
+        # OpenSRE CSV telemetry carries no deploy/config-change annotations.
+        return []
+
     def _default_alert(self) -> dict[str, Any]:
         return {
             "title": "OpenSRE local telemetry",

app/services/grafana/base.py

@@ -5,6 +5,7 @@
 import base64
 import json
 import logging
+from datetime import UTC, datetime
 from typing import Any
 from urllib.parse import quote
 
@@ -45,6 +46,29 @@ def _extract_rule_queries(rule: dict) -> list[dict]:
     return queries
 
 
+def _epoch_ms_to_iso(ms: Any) -> str | None:
+    """Convert a Grafana epoch-millisecond timestamp to an ISO 8601 UTC string."""
+    if ms is None:
+        return None
+    try:
+        return datetime.fromtimestamp(int(ms) / 1000, tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+    except (TypeError, ValueError, OSError):
+        return None
+
+
+def _map_annotation(item: dict[str, Any]) -> dict[str, Any]:
+    """Map a raw /api/annotations item to the tool-facing annotation shape."""
+    tags = item.get("tags")
+    return {
+        "time": _epoch_ms_to_iso(item.get("time")),
+        "time_end": _epoch_ms_to_iso(item.get("timeEnd")),
+        "text": item.get("text", ""),
+        "tags": tags if isinstance(tags, list) else [],
+        # Modern UID only; ignore the legacy numeric dashboardId (0 == not attached).
+        "dashboard_uid": item.get("dashboardUID") or None,
+    }
+
+
 class GrafanaClientBase:
     """Base HTTP client with common request methods for Grafana Cloud."""
 
@@ -295,6 +319,42 @@ def query_alert_rules(self, folder: str | None = None) -> list[dict[str, Any]]:
             logger.warning("[grafana] Failed to query alert rules: %s", e)
             return []
 
+    def query_annotations(
+        self,
+        from_ts: int,
+        to_ts: int,
+        tags: list[str] | None = None,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        """Query Grafana annotations in a time window (epoch ms), optional tag filter.
+
+        Mirrors ``query_alert_rules``: a direct ``requests.get`` returning a list.
+        ``/api/annotations`` responds with a JSON array, so ``_make_request`` (which
+        returns a dict) is unsuitable here.
+        """
+        url = f"{self.instance_url}/api/annotations"
+        params: dict[str, Any] = {
+            "from": from_ts,
+            "to": to_ts,
+            "type": "annotation",
+            "limit": limit,
+        }
+        if tags:
+            params["tags"] = tags  # requests repeats the param once per tag
+        try:
+            response = requests.get(
+                url,
+                headers=self._get_auth_headers(),
+                params=params,
+                timeout=10,
+            )
+            response.raise_for_status()
+            data = response.json()
+            return [_map_annotation(item) for item in data if isinstance(item, dict)]
+        except Exception as e:
+            logger.warning("[grafana] Failed to query annotations: %s", e)
+            return []
+
     def _get_auth_headers(self) -> dict[str, str]:
         if self.username and self.password:
             credentials = base64.b64encode(f"{self.username}:{self.password}".encode()).decode()

app/tools/GrafanaAnnotationsTool/__init__.py

@@ -0,0 +1,149 @@
+"""Grafana deployment-annotations query tool for change correlation."""
+
+from __future__ import annotations
+
+import time
+from datetime import UTC, datetime
+from typing import Any
+
+from app.services.grafana.base import _epoch_ms_to_iso, _map_annotation
+from app.tools.GrafanaLogsTool import (
+    _grafana_available,
+    _grafana_creds,
+    _grafana_source,
+    _resolve_grafana_client,
+)
+from app.tools.tool_decorator import tool
+
+
+def _query_grafana_annotations_extract_params(sources: dict[str, dict]) -> dict[str, Any]:
+    grafana = _grafana_source(sources)
+    return {
+        "time_range_minutes": grafana.get("time_range_minutes", 60),
+        "grafana_backend": grafana.get("_backend"),
+        **_grafana_creds(grafana),
+    }
+
+
+def _query_grafana_annotations_available(sources: dict[str, dict]) -> bool:
+    return _grafana_available(sources)
+
+
+def _normalize_backend_annotations(raw: Any) -> list[dict[str, Any]]:
+    """Normalize fixture/backend ``/api/annotations`` arrays to the client shape."""
+    if not isinstance(raw, list):
+        return []
+    return [_map_annotation(item) for item in raw if isinstance(item, dict)]
+
+
+def _iso_to_epoch_ms(value: str) -> int:
+    """Parse an ISO 8601 timestamp to epoch milliseconds (UTC). Raises ValueError if invalid.
+
+    A timezone-naive value (no ``Z`` / offset) is interpreted as UTC, not host-local time.
+    """
+    dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return int(dt.timestamp() * 1000)
+
+
+@tool(
+    name="query_grafana_annotations",
+    display_name="Grafana annotations",
+    source="grafana",
+    description=(
+        "Query Grafana deployment/config-change annotations to correlate changes with "
+        "an incident — the source-agnostic 'what changed and when' marker."
+    ),
+    use_cases=[
+        "Checking whether a deploy or config change preceded an alert",
+        "Correlating incidents with ArgoCD/Flux/Helm/Terraform/manual changes emitted as annotations",
+        "Building a source-agnostic change timeline alongside the GitHub deploy timeline",
+    ],
+    requires=[],
+    input_schema={
+        "type": "object",
+        "properties": {
+            "from": {
+                "type": "string",
+                "description": "ISO 8601 window start (overrides time_range_minutes)",
+            },
+            "to": {
+                "type": "string",
+                "description": "ISO 8601 window end (overrides time_range_minutes)",
+            },
+            "tags": {"type": "array", "items": {"type": "string"}},
+            "time_range_minutes": {"type": "integer", "default": 60},
+            "limit": {"type": "integer", "default": 100},
+            "grafana_endpoint": {"type": "string"},
+            "grafana_api_key": {"type": "string"},
+        },
+        "required": [],
+    },
+    is_available=_query_grafana_annotations_available,
+    extract_params=_query_grafana_annotations_extract_params,
+)
+def query_grafana_annotations(
+    tags: list[str] | None = None,
+    time_range_minutes: int = 60,
+    limit: int = 100,
+    grafana_endpoint: str | None = None,
+    grafana_api_key: str | None = None,
+    grafana_username: str = "",
+    grafana_password: str = "",
+    grafana_backend: Any = None,
+    **_kwargs: Any,
+) -> dict:
+    """Query Grafana annotations to correlate deploys/config changes with an incident.
+
+    ``from``/``to`` are accepted via the schema (ISO 8601); they are read from
+    ``_kwargs`` because ``from`` is a Python keyword and cannot be a parameter name.
+    When absent, the window defaults to the last ``time_range_minutes``.
+    """
+    if grafana_backend is not None:
+        raw = grafana_backend.query_annotations(tags=tags, limit=limit)
+        annotations = _normalize_backend_annotations(raw)
+        return {
+            "source": "grafana_annotations",
+            "available": True,
+            "annotations": annotations,
+            "total": len(annotations),
+            "raw": raw,
+        }
+
+    client = _resolve_grafana_client(
+        grafana_endpoint, grafana_api_key, grafana_username, grafana_password
+    )
+    if not client or not client.is_configured:
+        return {
+            "source": "grafana_annotations",
+            "available": False,
+            "error": "Grafana integration not configured",
+            "annotations": [],
+        }
+
+    now_ms = int(time.time() * 1000)
+    try:
+        from_iso, to_iso = _kwargs.get("from"), _kwargs.get("to")
+        to_ts = _iso_to_epoch_ms(to_iso) if to_iso else now_ms
+        # Default the window to end at `to` (now if unset), so a `to`-only call still
+        # yields a valid [to - window, to] range rather than from_ts > to_ts.
+        from_ts = _iso_to_epoch_ms(from_iso) if from_iso else to_ts - time_range_minutes * 60 * 1000
+    except (ValueError, TypeError, AttributeError) as e:
+        return {
+            "source": "grafana_annotations",
+            "available": False,
+            "error": f"Invalid timestamp: {e}",
+            "annotations": [],
+        }
+
+    annotations = client.query_annotations(from_ts=from_ts, to_ts=to_ts, tags=tags, limit=limit)
+    return {
+        "source": "grafana_annotations",
+        "available": True,
+        "annotations": annotations,
+        "total": len(annotations),
+        "tags_filter": tags,
+        "from": _epoch_ms_to_iso(from_ts),
+        "to": _epoch_ms_to_iso(to_ts),
+    }

docs/docs.json

@@ -108,6 +108,7 @@
               "coralogix",
               "datadog",
               "grafana",
+              "grafana_annotations",
               "hermes",
               "honeycomb",
               "incident_io",

docs/grafana_annotations.mdx

@@ -0,0 +1,71 @@
+---
+title: "Grafana Annotations"
+---
+
+Correlate incidents with **deployments and config changes** from any source. OpenSRE reads
+[Grafana annotations](https://grafana.com/docs/grafana/latest/dashboards/build-dashboards/annotate-visualizations/) —
+the standard, source-agnostic "what changed and when" marker — so the agent can answer
+_"did a deploy or config change precede this alert?"_ even when the change did **not** come
+from a GitHub push (ArgoCD/Flux syncs, `helm upgrade`, Jenkins/CircleCI jobs, Terraform
+applies, manual hotfixes).
+
+This complements the GitHub deploy timeline, which only sees GitHub-originated deploys.
+
+## Requirements
+
+No new setup or credentials — it reuses your existing **Grafana** integration. If Grafana is
+connected (see [Grafana](/grafana)), the annotations tool is available automatically during
+investigations.
+
+## Parameters
+
+| Parameter            | Description                                                                 |
+| -------------------- | --------------------------------------------------------------------------- |
+| `from`               | ISO 8601 window start (e.g. `2026-05-30T14:00:00Z`). Overrides the default. |
+| `to`                 | ISO 8601 window end. Overrides the default.                                 |
+| `tags`               | Optional list of annotation tags to filter by (e.g. `["deployment"]`).      |
+| `time_range_minutes` | Window size when `from`/`to` are omitted (default `60`, ending now).        |
+| `limit`              | Maximum annotations to return (default `100`).                              |
+
+## Example
+
+```text
+query_grafana_annotations(from="2026-05-30T14:00:00Z", to="2026-05-30T15:00:00Z", tags=["deployment"])
+```
+
+```json
+{
+  "source": "grafana_annotations",
+  "total": 1,
+  "annotations": [
+    {
+      "time": "2026-05-30T14:41:09Z",
+      "time_end": null,
+      "text": "deploy checkout-api v2.8.1",
+      "tags": ["deployment", "checkout-api"],
+      "dashboard_uid": null
+    }
+  ]
+}
+```
+
+Each annotation also carries `time_end` (set only for region annotations) and `dashboard_uid`
+(set only when the annotation is attached to a dashboard); both are `null` otherwise.
+
+The agent uses results like this to flag a likely change-induced regression and tie the
+suspected root cause to a specific deploy.
+
+## Emitting annotations
+
+Make your deploys visible by writing a Grafana annotation when you ship. Most CD tools can
+post to Grafana's annotations API, for example:
+
+```bash
+curl -s -X POST "$GRAFANA_URL/api/annotations" \
+  -H "Authorization: Bearer $GRAFANA_TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{"time": 1717079669000, "text": "deploy checkout-api v2.8.1", "tags": ["deployment", "checkout-api"]}'
+```
+
+Tagging deploy annotations consistently (e.g. `deployment`) lets the agent filter precisely
+during an investigation.

tests/synthetic/mock_grafana_backend/backend.py

@@ -34,11 +34,12 @@
 class GrafanaBackend(Protocol):
     """Minimal observability interface used by the RDS investigation agent.
 
-    Four methods — one per evidence pillar:
+    Four evidence pillars plus change correlation:
         query_timeseries  → Mimir/Prometheus matrix response
         query_logs        → Loki streams response
         query_alert_rules → Grafana Ruler rules response
         query_traces      → Tempo search response
+        query_annotations → /api/annotations array (deploy/config-change markers)
     """
 
     def query_timeseries(self, query: str = "", **kwargs: Any) -> dict[str, Any]:
@@ -57,6 +58,10 @@ def query_traces(self, **kwargs: Any) -> dict[str, Any]:
         """Return a Tempo-compatible search response."""
         pass
 
+    def query_annotations(self, **kwargs: Any) -> list[dict[str, Any]]:
+        """Return a Grafana /api/annotations array (a JSON list, not a dict)."""
+        pass
+
 
 class FixtureGrafanaBackend:
     """GrafanaBackend implementation backed by a ScenarioFixture.
@@ -181,3 +186,7 @@ def query_alert_rules(self, **_: Any) -> dict[str, Any]:
 
     def query_traces(self, **_: Any) -> dict[str, Any]:
         return format_tempo_search()
+
+    def query_annotations(self, **_: Any) -> list[dict[str, Any]]:
+        # RDS scenario fixtures declare no deploy/config-change annotations.
+        return []

tests/synthetic/mock_grafana_backend/selective_backend.py

@@ -92,6 +92,9 @@ def query_alert_rules(self, **_: Any) -> dict[str, Any]:
     def query_traces(self, **_: Any) -> dict[str, Any]:
         return format_tempo_search()
 
+    def query_annotations(self, **_: Any) -> list[dict[str, Any]]:
+        return []
+
     def reset(self) -> None:
         """Clear the queried_metrics audit log (useful for re-running a scenario)."""
         self.queried_metrics = []