Coverage for .github / scripts / validate_grafana_dashboards.py: 100.00%
161 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1#!/usr/bin/env python3
2"""Validate the curated Grafana dashboards against the Grafana we actually ship.
4The GCO dashboards live as JSON payloads inside ConfigMap manifests under
5``lambda/kubectl-applier-simple/manifests/`` and are imported at runtime by
6the kube-prometheus-stack Grafana sidecar (label ``grafana_dashboard="1"``).
7Nothing in that pipeline rejects a malformed or schema-incompatible dashboard
8loudly — it just fails to appear in the UI. This script closes that gap in CI
9by provisioning the extracted dashboards into the exact Grafana image the
10pinned chart version bundles and asserting each one loads.
12Subcommands:
14``extract``
15 Pull every dashboard JSON out of the given ConfigMap manifests, resolving
16 ``{{UPPER_SNAKE}}`` feature placeholders first (any value works — only
17 resolution matters here; the applier's substitution fidelity is covered by
18 ``tests/test_grafana_dashboards.py`` through the real handler). Writes
19 ``dashboards/<uid>.json`` plus a file-provisioning provider under
20 ``provisioning/`` shaped exactly like the sidecar's load path.
22``chart-version``
23 Print the pinned ``kube-prometheus-stack`` chart version and repo URL from
24 ``lambda/helm-installer/charts.yaml``, so the workflow resolves the
25 Grafana image from the same pin the deployment uses (no second pin to
26 drift). Appends ``version=``/``repo_url=`` lines to ``$GITHUB_OUTPUT``
27 when set.
29``verify``
30 Wait for a running Grafana's ``/api/health``, then assert every extracted
31 dashboard round-trips: ``GET /api/dashboards/uid/<uid>`` answers 200, the
32 title matches the source, and ``meta.provisioned`` is true.
34Importable (``extract_dashboards()``, ``read_chart_pin()``, ``verify()``) so
35pytest can hold the extraction in lockstep with the applier-path tests.
36"""
38from __future__ import annotations
40import argparse
41import base64
42import http.client
43import json
44import os
45import re
46import sys
47import time
48import urllib.error
49import urllib.request
50from pathlib import Path
51from typing import Any
53import yaml
55# Matches the applier's feature-gate regex: deliberately UPPER_SNAKE only, so
56# Grafana's own lowercase legend tokens ({{gpu}}, {{namespace}}, {{Hostname}})
57# pass through untouched.
58_FEATURE_PLACEHOLDER_RE = re.compile(r"\{\{[A-Z0-9_]+\}\}")
60_SIDECAR_LABEL = "grafana_dashboard"
62_PROVIDER_YAML = """\
63apiVersion: 1
64providers:
65 - name: gco-dashboards
66 type: file
67 updateIntervalSeconds: 10
68 options:
69 path: /var/lib/grafana/dashboards
70"""
73class ValidationError(Exception):
74 """A dashboard payload or Grafana response failed validation."""
77def _resolve_placeholders(content: str) -> str:
78 """Resolve every UPPER_SNAKE feature placeholder so YAML parses."""
79 return _FEATURE_PLACEHOLDER_RE.sub("true", content)
82def extract_dashboards(manifest_paths: list[Path]) -> dict[str, dict[str, Any]]:
83 """Return ``{uid: dashboard}`` for every sidecar-labeled ConfigMap payload.
85 Raises :class:`ValidationError` for anything the Grafana sidecar would
86 swallow silently: unparseable JSON, a missing/duplicate ``uid``, or a
87 missing ``title``.
88 """
89 dashboards: dict[str, dict[str, Any]] = {}
90 for path in manifest_paths:
91 content = _resolve_placeholders(path.read_text(encoding="utf-8"))
92 for document in yaml.safe_load_all(content):
93 if not isinstance(document, dict) or document.get("kind") != "ConfigMap":
94 continue
95 metadata = document.get("metadata") or {}
96 labels = metadata.get("labels") or {}
97 if str(labels.get(_SIDECAR_LABEL, "")) != "1":
98 continue
99 name = metadata.get("name", "<unnamed>")
100 for key, payload in (document.get("data") or {}).items():
101 if not str(key).endswith(".json"):
102 continue
103 where = f"{path.name} ConfigMap {name} data {key}"
104 try:
105 dashboard = json.loads(payload)
106 except json.JSONDecodeError as exc:
107 raise ValidationError(f"{where}: invalid JSON: {exc}") from exc
108 uid = dashboard.get("uid")
109 if not isinstance(uid, str) or not uid:
110 raise ValidationError(f"{where}: dashboard has no uid")
111 if uid in dashboards:
112 raise ValidationError(f"{where}: duplicate dashboard uid {uid!r}")
113 if not dashboard.get("title"):
114 raise ValidationError(f"{where}: dashboard has no title")
115 dashboards[uid] = dashboard
116 if not dashboards:
117 raise ValidationError("no sidecar-labeled dashboard ConfigMaps found")
118 return dashboards
121def read_chart_pin(charts_yaml: Path, chart: str = "kube-prometheus-stack") -> dict[str, str]:
122 """Read the pinned version and repo URL for ``chart`` from charts.yaml."""
123 data = yaml.safe_load(charts_yaml.read_text(encoding="utf-8"))
124 entries = data.get("charts", data) if isinstance(data, dict) else {}
125 for entry in entries.values():
126 if isinstance(entry, dict) and entry.get("chart") == chart:
127 version = str(entry.get("version", "")).strip()
128 repo_url = str(entry.get("repo_url", "")).strip()
129 if not version or not repo_url:
130 raise ValidationError(f"{chart} entry is missing version or repo_url")
131 return {"version": version, "repo_url": repo_url}
132 raise ValidationError(f"no {chart} entry found in {charts_yaml}")
135# Polling a Grafana that is still booting sees the whole zoo of transport
136# failures: connection refused, docker-proxy accepting then resetting the
137# socket (raw ConnectionResetError, not wrapped in URLError), and half-open
138# responses (http.client.RemoteDisconnected). OSError covers URLError,
139# TimeoutError, and every Connection*Error; HTTPException covers the
140# half-open cases. urllib.error.HTTPError is caught separately first, so
141# real HTTP status codes are still returned rather than swallowed here.
142_RETRIABLE_FETCH_ERRORS = (OSError, http.client.HTTPException, json.JSONDecodeError)
145def _get(url: str, auth: tuple[str, str] | None = None, timeout: float = 10.0) -> tuple[int, Any]:
146 """GET a Grafana API URL, returning (status, parsed JSON or None)."""
147 if not url.startswith(("http://", "https://")):
148 raise ValidationError(f"refusing non-HTTP URL {url!r}")
149 request = urllib.request.Request(url) # noqa: S310 - scheme validated above
150 if auth is not None:
151 token = base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode()
152 request.add_header("Authorization", f"Basic {token}")
153 try:
154 # The URL is assembled from the --url flag this CI job passes for its
155 # own localhost container plus repo-controlled dashboard uids, and the
156 # scheme check above rejects anything that is not plain HTTP(S).
157 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
158 with urllib.request.urlopen(request, timeout=timeout) as response: # nosec B310
159 return response.status, json.loads(response.read().decode("utf-8"))
160 except urllib.error.HTTPError as exc:
161 return exc.code, None
162 except _RETRIABLE_FETCH_ERRORS:
163 return 0, None
166def verify(
167 url: str,
168 dashboards_dir: Path,
169 user: str,
170 password: str,
171 timeout_seconds: float = 180.0,
172) -> list[str]:
173 """Assert every extracted dashboard is provisioned in the running Grafana.
175 Returns a list of failure messages (empty when everything passed).
176 """
177 deadline = time.monotonic() + timeout_seconds
178 while True:
179 status, health = _get(f"{url}/api/health")
180 if status == 200 and isinstance(health, dict) and health.get("database") == "ok":
181 print(f"Grafana healthy: version {health.get('version', '<unknown>')}")
182 break
183 if time.monotonic() >= deadline:
184 return [f"Grafana at {url} did not become healthy within {timeout_seconds:.0f}s"]
185 time.sleep(2)
187 sources = sorted(dashboards_dir.glob("*.json"))
188 if not sources:
189 return [f"no extracted dashboards found under {dashboards_dir}"]
191 failures: list[str] = []
192 for source_path in sources:
193 source = json.loads(source_path.read_text(encoding="utf-8"))
194 uid, title = source["uid"], source["title"]
195 # File provisioning is asynchronous; give the provisioner a moment
196 # per dashboard rather than one global sleep.
197 item_deadline = time.monotonic() + 30
198 while True:
199 status, body = _get(f"{url}/api/dashboards/uid/{uid}", auth=(user, password))
200 if status == 200 or time.monotonic() >= item_deadline:
201 break
202 time.sleep(2)
203 if status != 200 or not isinstance(body, dict):
204 failures.append(f"{uid}: Grafana answered {status}, expected 200")
205 continue
206 loaded_title = (body.get("dashboard") or {}).get("title")
207 provisioned = (body.get("meta") or {}).get("provisioned")
208 if loaded_title != title:
209 failures.append(f"{uid}: loaded title {loaded_title!r} != source {title!r}")
210 elif provisioned is not True:
211 failures.append(f"{uid}: dashboard loaded but meta.provisioned is {provisioned!r}")
212 else:
213 print(f"PASS {uid}: provisioned as {title!r}")
214 return failures
217def _cmd_extract(args: argparse.Namespace) -> int:
218 dashboards = extract_dashboards([Path(item) for item in args.manifest])
219 out_dir = Path(args.out_dir)
220 dashboards_dir = out_dir / "dashboards"
221 provisioning_dir = out_dir / "provisioning"
222 dashboards_dir.mkdir(parents=True, exist_ok=True)
223 provisioning_dir.mkdir(parents=True, exist_ok=True)
224 (provisioning_dir / "gco-dashboards.yaml").write_text(_PROVIDER_YAML, encoding="utf-8")
225 for uid, dashboard in sorted(dashboards.items()):
226 target = dashboards_dir / f"{uid}.json"
227 target.write_text(json.dumps(dashboard, indent=2), encoding="utf-8")
228 print(f"extracted uid={uid} title={dashboard['title']!r} -> {target}")
229 print(f"extracted {len(dashboards)} dashboard(s)")
230 return 0
233def _cmd_chart_version(args: argparse.Namespace) -> int:
234 pin = read_chart_pin(Path(args.charts_yaml))
235 lines = [f"version={pin['version']}", f"repo_url={pin['repo_url']}"]
236 for line in lines:
237 print(line)
238 github_output = os.environ.get("GITHUB_OUTPUT")
239 if github_output:
240 with open(github_output, "a", encoding="utf-8") as handle:
241 handle.write("\n".join(lines) + "\n")
242 return 0
245def _cmd_verify(args: argparse.Namespace) -> int:
246 failures = verify(
247 args.url.rstrip("/"),
248 Path(args.dashboards_dir),
249 user=os.environ.get("GRAFANA_USER", "admin"),
250 password=os.environ.get("GRAFANA_PASSWORD", "admin"),
251 timeout_seconds=args.timeout,
252 )
253 for failure in failures:
254 print(f"FAIL {failure}", file=sys.stderr)
255 return 1 if failures else 0
258def main(argv: list[str] | None = None) -> int:
259 parser = argparse.ArgumentParser(description=__doc__)
260 subparsers = parser.add_subparsers(dest="command", required=True)
262 extract = subparsers.add_parser("extract", help="extract dashboard JSONs from manifests")
263 extract.add_argument("--manifest", action="append", required=True)
264 extract.add_argument("--out-dir", required=True)
265 extract.set_defaults(func=_cmd_extract)
267 chart = subparsers.add_parser("chart-version", help="print the pinned chart version")
268 chart.add_argument(
269 "--charts-yaml",
270 default="lambda/helm-installer/charts.yaml",
271 )
272 chart.set_defaults(func=_cmd_chart_version)
274 check = subparsers.add_parser("verify", help="assert dashboards provisioned in Grafana")
275 check.add_argument("--url", default="http://127.0.0.1:3000")
276 check.add_argument("--dashboards-dir", required=True)
277 check.add_argument("--timeout", type=float, default=180.0)
278 check.set_defaults(func=_cmd_verify)
280 args = parser.parse_args(argv)
281 try:
282 return int(args.func(args))
283 except ValidationError as exc:
284 print(f"FAIL {exc}", file=sys.stderr)
285 return 1
288if __name__ == "__main__":
289 sys.exit(main())