#!/usr/bin/env python3 """ Self-updating sync tool for the cloudnative-pg Grafana dashboard. Designed for unattended / background-agent runs: fetches the upstream dashboard JSON, applies the transformations declared in `monitoring/cnpg-dashboard-transforms.yaml`, writes the result to `knoe-db/grafana-dashboard.json` (the canonical source the `knoe-db-grafana-dashboard` ConfigMap is built from), and optionally applies the ConfigMap to a live cluster + reloads Grafana provisioning. Usage: # Default: fetch upstream, apply transforms, write to disk if changed. # Exit 0 = no change, 1 = updated, 2 = error. ./etc/sync_cnpg_grafana_dashboard.py # Dry run — print what would change, don't write anything. ./etc/sync_cnpg_grafana_dashboard.py --dry-run # Drift check — exit 0 if our committed copy matches the upstream-after- # transforms, 1 if it has drifted (newer upstream waiting to be merged # OR a local hand-edit that the transforms don't cover). Doesn't write. ./etc/sync_cnpg_grafana_dashboard.py --check # Apply to live cluster after writing (rolls Grafana provisioning). ./etc/sync_cnpg_grafana_dashboard.py --apply --context $APP_CLUSTER_KUBECONTEXT Exit codes: 0 no change (output matches transformed upstream) 1 output updated (or would-be-updated in --dry-run / --check) 2 error (network, parse, transform mismatch, kubectl failure, etc.) Background-agent run pattern (cron / k8s CronJob): 1. Run with `--check` once. If exit 1, work to do. 2. Run with `--apply --context=` to update + roll out. 3. Optionally `git commit -m 'sync(cnpg): ...'` to persist the file change. Transformation philosophy: edits live in `monitoring/cnpg-dashboard-transforms.yaml`, not in this tool. Adding a new transformation type means extending `Transformer.apply()` here AND documenting it in the transforms YAML's preamble. Today there's just one type, `regex_replace`. """ from __future__ import annotations import argparse import base64 import difflib import hashlib import json import logging import os import re import subprocess import sys import tempfile import urllib.error import urllib.request from pathlib import Path try: import yaml except ImportError: print( "PyYAML required. Install via: pip install pyyaml (or apt install python3-yaml)", file=sys.stderr, ) sys.exit(2) # -------------------------------------------------------------------------- # Constants REPO_ROOT = Path(__file__).resolve().parent.parent DEFAULT_TRANSFORMS = REPO_ROOT / "monitoring" / "cnpg-dashboard-transforms.yaml" DEFAULT_OUTPUT = REPO_ROOT / "knoe-db" / "grafana-dashboard.json" DEFAULT_CM_NAME = "knoe-db-grafana-dashboard" DEFAULT_GRAFANA_STS = "kps-grafana" DEFAULT_GRAFANA_SECRET = "kps-grafana" EXIT_NOCHANGE = 0 EXIT_CHANGED = 1 EXIT_ERROR = 2 logger = logging.getLogger("cnpg-dashboard-sync") # -------------------------------------------------------------------------- # Helpers def sha256_short(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest()[:12] def fetch_upstream(url: str, timeout: int = 30) -> str: """Download the upstream dashboard JSON. Raises urllib.error on HTTP failure.""" logger.info("fetching upstream: %s", url) req = urllib.request.Request(url, headers={"User-Agent": "knoe-cnpg-dashboard-sync/1"}) with urllib.request.urlopen(req, timeout=timeout) as r: return r.read().decode("utf-8") def load_transforms(path: Path) -> dict: if not path.exists(): raise FileNotFoundError(f"transforms file not found: {path}") with path.open() as f: data = yaml.safe_load(f) if not isinstance(data, dict) or "source" not in data: raise ValueError(f"transforms file missing required `source` block: {path}") return data # -------------------------------------------------------------------------- # Transformer class TransformError(Exception): pass class Transformer: """Applies an ordered list of transformations to a JSON string.""" def __init__(self, transforms: list[dict]): self.transforms = transforms def apply(self, content: str) -> tuple[str, list[dict]]: """Apply all transforms; return (transformed_content, summaries). Each summary is a dict {name, type, matches, ...} for logging / drift detection. """ summaries: list[dict] = [] for t in self.transforms: name = t.get("name", "") ttype = t.get("type") try: if ttype == "regex_replace": content, summary = self._apply_regex(content, t) elif ttype == "set_template_variable_default": content, summary = self._apply_set_template_default(content, t) else: raise TransformError(f"unknown transformation type: {ttype!r}") except TransformError: raise except Exception as e: raise TransformError(f"transform {name!r} failed: {e}") from e summary["name"] = name summary["type"] = ttype summaries.append(summary) return content, summaries @staticmethod def _apply_regex(content: str, t: dict) -> tuple[str, dict]: pattern = t["pattern"] replacement = t["replacement"] compiled = re.compile(pattern) new_content, n = compiled.subn(replacement, content) emin = t.get("expected_min") emax = t.get("expected_max") warn = None if emin is not None and n < emin: warn = f"matched {n} < expected_min {emin} — upstream may have moved on" elif emax is not None and n > emax: warn = f"matched {n} > expected_max {emax} — upstream changed shape" return new_content, {"matches": n, "warning": warn} @staticmethod def _apply_set_template_default(content: str, t: dict) -> tuple[str, dict]: """Set the `current` value of a template variable. Useful for pinning DS_PROMETHEUS (or any other variable) to a specific value so the dashboard renders correctly on first load without the user needing to override via URL parameters or the variable picker. Spec: type: set_template_variable_default variable: DS_PROMETHEUS # required, name of variable value: cnpg-prometheus # required, current.value text: cnpg-prometheus # optional, defaults to value selected: true # optional, defaults to true """ var_name = t["variable"] value = t["value"] text = t.get("text", value) selected = t.get("selected", True) data = json.loads(content) templating = data.get("templating", {}).get("list", []) found = False for v in templating: if v.get("name") == var_name: v["current"] = {"selected": selected, "text": text, "value": value} found = True break if not found: warn = f"variable {var_name!r} not found in dashboard.templating.list" return content, {"matches": 0, "warning": warn} # Re-serialize with the same format the upstream JSON uses (2-space indent, # which is what `json.dumps(..., indent=2)` produces; matches what we get # from the upstream so diffs stay clean). return json.dumps(data, indent=2), {"matches": 1, "warning": None} # -------------------------------------------------------------------------- # Cluster apply def kubectl(args: list[str], context: str, namespace: str | None = None, stdin: bytes | None = None, capture: bool = False) -> bytes: """Run kubectl with explicit context (no ambient context drift).""" cmd = ["kubectl", "--context", context] if namespace: cmd += ["-n", namespace] cmd += args logger.debug("running: %s", " ".join(cmd)) if capture: return subprocess.check_output(cmd, input=stdin) subprocess.run(cmd, input=stdin, check=True) return b"" def apply_to_cluster(json_content: str, context: str, namespace: str = "monitoring", cm_name: str = DEFAULT_CM_NAME) -> None: """Apply the dashboard JSON as a ConfigMap + reload Grafana provisioning. Uses --server-side apply because the JSON is large enough to hit the 256KB last-applied-configuration annotation limit on client-side apply. """ if not context: raise ValueError("--apply requires --context (or APP_CLUSTER_KUBECONTEXT env)") logger.info("applying ConfigMap %s/%s in context %s", namespace, cm_name, context) with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: f.write(json_content) tmp = f.name try: # Render the cm yaml via `kubectl create --dry-run` cm_yaml = kubectl( ["create", "configmap", cm_name, f"--from-file=knoe-db.json={tmp}", "--dry-run=client", "-o", "yaml"], context=context, namespace=namespace, capture=True, ) # Apply server-side kubectl( ["apply", "--server-side", "--force-conflicts", "--field-manager=cnpg-dashboard-sync", "-f", "-"], context=context, namespace=namespace, stdin=cm_yaml, ) # Sidecar discovers grafana_dashboard=1 labelled cms; ensure label is present kubectl( ["label", "configmap", cm_name, "grafana_dashboard=1", "--overwrite"], context=context, namespace=namespace, ) finally: os.unlink(tmp) # Reload Grafana provisioning so the new dashboard JSON is picked up # without waiting for the sidecar's poll interval. logger.info("reloading Grafana provisioning") admin_pw_b64 = kubectl( ["get", "secret", DEFAULT_GRAFANA_SECRET, "-o", "jsonpath={.data.admin-password}"], context=context, namespace=namespace, capture=True, ) admin_pw = base64.b64decode(admin_pw_b64).decode() kubectl( ["exec", f"sts/{DEFAULT_GRAFANA_STS}", "-c", "grafana", "--", "wget", "-qO-", "--post-data", "", f"http://admin:{admin_pw}@localhost:3000/api/admin/provisioning/dashboards/reload"], context=context, namespace=namespace, capture=True, ) logger.info("grafana provisioning reload complete") # -------------------------------------------------------------------------- # Main def render_diff(old: str, new: str, max_lines: int = 60) -> str: diff = list(difflib.unified_diff( old.splitlines(keepends=True), new.splitlines(keepends=True), fromfile="committed", tofile="upstream-transformed", n=2, )) if len(diff) > max_lines: diff = diff[:max_lines] + [f"... ({len(diff) - max_lines} more lines elided)\n"] return "".join(diff) def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument("--transforms", type=Path, default=DEFAULT_TRANSFORMS, help="Path to the transforms YAML (default: %(default)s)") p.add_argument("--output", type=Path, default=DEFAULT_OUTPUT, help="Path to write the transformed JSON (default: %(default)s)") mode = p.add_mutually_exclusive_group() mode.add_argument("--dry-run", action="store_true", help="Show what would change; don't write or apply.") mode.add_argument("--check", action="store_true", help="Exit 1 if output is drifted from upstream-transformed; don't write.") p.add_argument("--apply", action="store_true", help="After writing, apply the ConfigMap to the live cluster and reload Grafana.") p.add_argument("--context", default=os.environ.get("APP_CLUSTER_KUBECONTEXT", ""), help="kubectl context for --apply (default: $APP_CLUSTER_KUBECONTEXT)") p.add_argument("--namespace", default="monitoring", help="ConfigMap namespace for --apply (default: %(default)s)") p.add_argument("--show-diff", action="store_true", help="Print a unified diff of committed-vs-transformed.") p.add_argument("-v", "--verbose", action="store_true", help="Verbose logging.") args = p.parse_args(argv) logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", datefmt="%H:%M:%S", ) try: config = load_transforms(args.transforms) except (FileNotFoundError, ValueError) as e: logger.error("failed to load transforms: %s", e) return EXIT_ERROR src = config["source"] upstream_url = src["url"] expected_uid = src.get("expected_uid") transforms = config.get("transformations", []) # 1. Fetch try: upstream = fetch_upstream(upstream_url) except urllib.error.URLError as e: logger.error("upstream fetch failed: %s", e) return EXIT_ERROR logger.info("upstream: sha256=%s, %d bytes", sha256_short(upstream), len(upstream)) # 2. Validate UID (catch silent renames) if expected_uid: try: uid = json.loads(upstream).get("uid") except json.JSONDecodeError as e: logger.error("upstream is not valid JSON: %s", e) return EXIT_ERROR if uid != expected_uid: logger.error("upstream uid %r != expected %r — refusing to proceed; " "dashboard URLs would silently break. Update transforms YAML if intentional.", uid, expected_uid) return EXIT_ERROR # 3. Transform try: transformed, summaries = Transformer(transforms).apply(upstream) except TransformError as e: logger.error("transform failed: %s", e) return EXIT_ERROR logger.info("applied %d transformation(s):", len(summaries)) saw_warning = False for s in summaries: msg = f" - {s['name']} ({s['type']}): {s.get('matches', '?')} match(es)" if s.get("warning"): msg += f" [WARNING: {s['warning']}]" saw_warning = True logger.info(msg) logger.info("transformed: sha256=%s, %d bytes", sha256_short(transformed), len(transformed)) # 4. Compare to existing existing = args.output.read_text() if args.output.exists() else "" drifted = (existing != transformed) if args.show_diff and drifted: logger.info("DIFF (committed vs transformed-upstream):") sys.stderr.write(render_diff(existing, transformed)) # 5. Act per mode if args.check: if drifted: logger.warning("DRIFT: %s differs from upstream after transforms (run without --check to update)", args.output) return EXIT_CHANGED logger.info("OK: %s matches upstream after transforms", args.output) return EXIT_ERROR if saw_warning else EXIT_NOCHANGE if args.dry_run: if drifted: logger.info("WOULD UPDATE: %s (run without --dry-run to apply)", args.output) return EXIT_CHANGED logger.info("NO CHANGE: %s", args.output) return EXIT_ERROR if saw_warning else EXIT_NOCHANGE # Default mode: write if drifted, optionally apply if drifted: args.output.write_text(transformed) logger.info("wrote %s", args.output) rc = EXIT_CHANGED else: logger.info("no change to %s", args.output) rc = EXIT_NOCHANGE if args.apply: try: apply_to_cluster(transformed, context=args.context, namespace=args.namespace) except (subprocess.CalledProcessError, ValueError) as e: logger.error("cluster apply failed: %s", e) return EXIT_ERROR return EXIT_ERROR if saw_warning else rc if __name__ == "__main__": sys.exit(main())