test(multiplayer): verify observability manifests locally

docs/OBSERVABILITY.md

@@ -7,6 +7,9 @@ the documented 250 ms p95 SLO with `histogram_quantile`. The optional
 5xx alerts for clusters running the Prometheus Operator. The matching optional
 `deploy/observability/prometheus-service-monitor.yaml` discovers the internal
 control-plane Service on its named `http` port and scrapes only `/metrics`.
+`scripts/verify_observability_manifests.py` is included in the local
+multiplayer gate and checks this Service/monitor contract without requiring a
+Kubernetes or Prometheus installation.
 
 Install the rule only after confirming that the `PrometheusRule` CRD and the
 `ServiceMonitor` CRD and the `cosmic-clash` namespace exist. The example

scripts/verify_multiplayer_local.sh

@@ -30,5 +30,6 @@ python3 "$root_dir/server/migrations/test_migration.py"
 python3 "$root_dir/server/security/test_fleet_manifests.py"
 python3 "$root_dir/server/security/test_kubernetes_policies.py"
 python3 "$root_dir/server/security/test_supply_chain.py"
+python3 "$root_dir/scripts/verify_observability_manifests.py"
 
 echo "LOCAL MULTIPLAYER GATE PASS"

scripts/verify_observability_manifests.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Validate the checked-in Prometheus discovery and alert resources."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import re
+import sys
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_DIRECTORY = ROOT / "deploy" / "observability"
+
+
+def verify(directory: Path, service_path: Path) -> None:
+    monitor = (directory / "prometheus-service-monitor.yaml").read_text()
+    rules = (directory / "prometheus-rules.yaml").read_text()
+    service = service_path.read_text()
+
+    if "kind: ServiceMonitor" not in monitor:
+        raise ValueError("ServiceMonitor resource is missing")
+    if "apiVersion: monitoring.coreos.com/v1" not in monitor:
+        raise ValueError("ServiceMonitor API version is not pinned")
+    if "namespace: cosmic-clash" not in monitor or '      - cosmic-clash' not in monitor:
+        raise ValueError("ServiceMonitor namespace is not restricted to cosmic-clash")
+    if "app.kubernetes.io/name: control-plane" not in monitor:
+        raise ValueError("ServiceMonitor does not select the control-plane")
+    if not re.search(r"(?m)^    - port: http$", monitor):
+        raise ValueError("ServiceMonitor does not use the named http port")
+    if not re.search(r"(?m)^      path: /metrics$", monitor):
+        raise ValueError("ServiceMonitor path is not /metrics")
+    if "interval: 15s" not in monitor or "scrapeTimeout: 5s" not in monitor:
+        raise ValueError("ServiceMonitor interval/timeout contract changed")
+
+    if "kind: Service" not in service or "name: control-plane" not in service:
+        raise ValueError("control-plane Service is missing")
+    if not re.search(r"(?m)^    - name: http$", service):
+        raise ValueError("control-plane Service has no named http port")
+
+    if "kind: PrometheusRule" not in rules:
+        raise ValueError("PrometheusRule resource is missing")
+    for alert in ("CosmicClashControlPlaneAPIP95High", "CosmicClashControlPlaneAPI5xxHigh"):
+        if f"alert: {alert}" not in rules:
+            raise ValueError(f"required alert is missing: {alert}")
+    if "histogram_quantile" not in rules or "cosmic_clash_api_latency_seconds_bucket" not in rules:
+        raise ValueError("API p95 alert is not based on the exported histogram")
+    if 'status="5xx"' not in rules or "cosmic_clash_api_requests_total" not in rules:
+        raise ValueError("API error alert is not based on the exported counter")
+    if "severity: page" not in rules or "owner: api" not in rules:
+        raise ValueError("alerts must have bounded routing labels")
+    routing = rules.split("labels:", 1)[-1].split("annotations:", 1)[0]
+    if "{{ $labels." in routing:
+        raise ValueError("dynamic labels were added to alert routing")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--directory", type=Path, default=DEFAULT_DIRECTORY)
+    parser.add_argument("--service", type=Path, default=ROOT / "deploy/k8s/base/control-plane-service.yaml")
+    args = parser.parse_args()
+    try:
+        verify(args.directory, args.service)
+    except (OSError, ValueError) as error:
+        print(f"observability manifest verification failed: {error}", file=sys.stderr)
+        return 1
+    print("observability manifest verification passed")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server/security/test_observability_manifests.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).parents[2]
+CHECKER = ROOT / "scripts" / "verify_observability_manifests.py"
+
+
+class ObservabilityManifestTest(unittest.TestCase):
+    def run_checker(self, directory=None):
+        command = [sys.executable, str(CHECKER)]
+        if directory is not None:
+            command += ["--directory", str(directory)]
+        return subprocess.run(command, cwd=ROOT, text=True, capture_output=True)
+
+    def test_checked_in_resources_match_service_and_metric_contract(self):
+        result = self.run_checker()
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_checker_rejects_wrong_namespace_and_broad_scrape(self):
+        with tempfile.TemporaryDirectory() as directory:
+            target = Path(directory)
+            for name in ("prometheus-service-monitor.yaml", "prometheus-rules.yaml"):
+                (target / name).write_text((ROOT / "deploy/observability" / name).read_text())
+            monitor = target / "prometheus-service-monitor.yaml"
+            monitor.write_text(monitor.read_text().replace("path: /metrics", "path: /").replace("- cosmic-clash", "- default"))
+            result = self.run_checker(target)
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn("namespace", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()