Coverage for scripts / example_job_validation / static_checks.py: 100.00%
175 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Offline validation of every example against its documented contract.
3Runs with no AWS access and no cluster: parses each example, checks the
4spec registry's symmetry with the ``examples/`` directory and the
5``gco_mcp`` catalog, and — for examples documented to travel the API/SQS
6submission paths — proves every document clears the exact transport gates
7(kind/GVK allowlist, image-source trust, target namespace) that the
8deployed services enforce. This is the half that runs in CI on every PR
9(``tests/test_example_job_validation.py``); the live half in
10``checks/examples.py`` builds on the same parse.
11"""
13from __future__ import annotations
15from dataclasses import dataclass, field
16from pathlib import Path
17from typing import Any
19import yaml
21from .specs import (
22 COMPANION,
23 DAG_RUN,
24 EXAMPLE_SPECS,
25 SUBMISSION_PATHS,
26 SUBMIT_API,
27 SUBMIT_DIRECT,
28 SUBMIT_SQS,
29 ExampleSpec,
30)
32#: Namespaces the platform provisions for user workloads.
33_WORKLOAD_NAMESPACES = frozenset({"gco-jobs", "gco-inference"})
36@dataclass
37class StaticFinding:
38 """One offline check outcome for one example."""
40 example: str
41 check: str
42 passed: bool
43 detail: str = ""
46@dataclass
47class ParsedExample:
48 """An example file parsed into documents plus its spec."""
50 name: str
51 path: Path
52 spec: ExampleSpec
53 documents: list[dict[str, Any]] = field(default_factory=list)
56def examples_dir(repo_root: Path) -> Path:
57 return repo_root / "examples"
60def example_names(repo_root: Path) -> list[str]:
61 return sorted(path.stem for path in examples_dir(repo_root).glob("*.yaml"))
64def parse_example(repo_root: Path, name: str) -> ParsedExample:
65 """Parse one example's YAML documents (raises on unknown name or bad YAML)."""
66 spec = EXAMPLE_SPECS.get(name)
67 if spec is None:
68 raise KeyError(f"No validation spec for example {name!r} (add one in specs.py)")
69 path = examples_dir(repo_root) / f"{name}.yaml"
70 documents = [
71 doc for doc in yaml.safe_load_all(path.read_text(encoding="utf-8")) if doc is not None
72 ]
73 return ParsedExample(name=name, path=path, spec=spec, documents=documents)
76def _catalog_metadata(repo_root: Path) -> dict[str, dict[str, Any]]:
77 """Read ``gco_mcp``'s EXAMPLE_METADATA literal without executing the module.
79 ``gco_mcp/resources/docs.py`` imports flat sibling modules (the MCP server
80 puts ``gco_mcp/`` itself on ``sys.path``), so importing it from here would
81 require path surgery. The catalog is a pure literal dict, so an AST read
82 is sufficient — and side-effect free.
83 """
84 import ast
86 docs_path = repo_root / "gco_mcp" / "resources" / "docs.py"
87 tree = ast.parse(docs_path.read_text(encoding="utf-8"))
88 for node in ast.walk(tree):
89 value: ast.expr | None = None
90 targets: list[ast.expr] = []
91 if isinstance(node, ast.Assign):
92 targets, value = node.targets, node.value
93 elif isinstance(node, ast.AnnAssign) and node.value is not None:
94 targets, value = [node.target], node.value
95 for target in targets:
96 if isinstance(target, ast.Name) and target.id == "EXAMPLE_METADATA":
97 assert value is not None
98 catalog = ast.literal_eval(value)
99 if not isinstance(catalog, dict):
100 raise RuntimeError("EXAMPLE_METADATA is not a dict literal")
101 return catalog
102 raise RuntimeError(f"EXAMPLE_METADATA literal not found in {docs_path}")
105def check_registry_symmetry(repo_root: Path) -> list[StaticFinding]:
106 """Specs, files, and the MCP catalog must describe the same example set."""
107 findings: list[StaticFinding] = []
108 files = set(example_names(repo_root))
109 specs = set(EXAMPLE_SPECS)
110 catalog = set(_catalog_metadata(repo_root))
111 findings.append(
112 StaticFinding(
113 example="*",
114 check="spec/file symmetry",
115 passed=files == specs,
116 detail=(
117 f"only in examples/: {sorted(files - specs)}; only in specs: {sorted(specs - files)}"
118 if files != specs
119 else ""
120 ),
121 )
122 )
123 findings.append(
124 StaticFinding(
125 example="*",
126 check="spec/catalog symmetry",
127 passed=catalog == specs,
128 detail=(
129 f"only in catalog: {sorted(catalog - specs)}; only in specs: {sorted(specs - catalog)}"
130 if catalog != specs
131 else ""
132 ),
133 )
134 )
135 return findings
138def check_submission_matches_catalog(repo_root: Path, name: str) -> StaticFinding:
139 """The spec's submission path must agree with the catalog's documented command."""
140 meta = _catalog_metadata(repo_root).get(name, {})
141 documented = str(meta.get("submission", ""))
142 spec = EXAMPLE_SPECS[name]
143 expectations = {
144 SUBMIT_DIRECT: "gco jobs submit-direct",
145 SUBMIT_SQS: "gco jobs submit-sqs",
146 SUBMIT_API: "gco jobs submit ",
147 DAG_RUN: "gco dag run",
148 }
149 if spec.submission == SUBMIT_DIRECT:
150 # Inference examples document `gco inference deploy` as the
151 # recommended path and manifest-direct submission as the alternative;
152 # both are valid documented shapes for a submit-direct spec.
153 ok = "gco jobs submit-direct" in documented or "gco inference deploy" in documented
154 detail = "" if ok else f"catalog documents {documented!r}, spec says {spec.submission}"
155 elif spec.submission in expectations:
156 ok = expectations[spec.submission] in documented
157 detail = "" if ok else f"catalog documents {documented!r}, spec says {spec.submission}"
158 elif spec.submission == COMPANION:
159 ok = True
160 detail = ""
161 else: # kubectl-apply
162 ok = "kubectl apply" in documented or documented == ""
163 detail = "" if ok else f"catalog documents {documented!r}, spec says kubectl-apply"
164 return StaticFinding(example=name, check="documented submission path", passed=ok, detail=detail)
167def check_transport_acceptance(parsed: ParsedExample) -> list[StaticFinding]:
168 """API/SQS-documented examples must clear the deployed validation gates."""
169 if parsed.spec.submission not in {SUBMIT_DIRECT, SUBMIT_SQS, SUBMIT_API}:
170 return []
171 # Module-level pure functions: no Kubernetes client construction, so the
172 # checks run on machines with no kubeconfig (CI runners, fresh laptops).
173 from gco.services.manifest_processor import validate_image_sources, validate_resource_kind
175 findings: list[StaticFinding] = []
176 for doc in parsed.documents:
177 label = f"{doc.get('kind')}/{(doc.get('metadata') or {}).get('name')}"
178 if parsed.spec.submission in {SUBMIT_SQS, SUBMIT_API}:
179 # Only the SQS/API services enforce the kind allowlist;
180 # submit-direct is client-side kubectl and takes any kind.
181 kind_ok, kind_reason = validate_resource_kind(doc)
182 findings.append(
183 StaticFinding(
184 example=parsed.name,
185 check=f"transport kind allowlist ({label})",
186 passed=kind_ok,
187 detail=kind_reason or "",
188 )
189 )
190 image_ok, image_reason = validate_image_sources(doc)
191 findings.append(
192 StaticFinding(
193 example=parsed.name,
194 check=f"trusted image sources ({label})",
195 passed=image_ok,
196 detail=image_reason or "",
197 )
198 )
199 return findings
202def check_namespaces(parsed: ParsedExample) -> list[StaticFinding]:
203 """Namespaced example documents must target a provisioned workload namespace."""
204 findings: list[StaticFinding] = []
205 for doc in parsed.documents:
206 metadata = doc.get("metadata") or {}
207 namespace = metadata.get("namespace")
208 kind = str(doc.get("kind", ""))
209 if kind in {"ResourceFlavor", "ClusterQueue"}: # cluster-scoped
210 continue
211 if namespace is None:
212 continue
213 findings.append(
214 StaticFinding(
215 example=parsed.name,
216 check=f"workload namespace ({kind}/{metadata.get('name')})",
217 passed=namespace in _WORKLOAD_NAMESPACES,
218 detail="" if namespace in _WORKLOAD_NAMESPACES else f"namespace {namespace!r}",
219 )
220 )
221 return findings
224def check_spec_shape(name: str) -> StaticFinding:
225 """Spec fields must use known enumerations."""
226 spec = EXAMPLE_SPECS[name]
227 ok = spec.submission in SUBMISSION_PATHS
228 return StaticFinding(
229 example=name,
230 check="spec shape",
231 passed=ok,
232 detail="" if ok else f"unknown submission path {spec.submission!r}",
233 )
236#: Container resource dimensions governed by the gco-jobs LimitRange, mapped
237#: to their per-container default ceiling key in DEFAULT_RESOURCE_QUOTA.
238_LIMIT_RANGE_DIMENSIONS = {
239 "cpu": "container_max_cpu",
240 "memory": "container_max_memory",
241 "nvidia.com/gpu": "container_max_gpu",
242}
244#: Aggregate request dimensions governed by the gco-jobs ResourceQuota.
245_QUOTA_DIMENSIONS = {
246 "cpu": "max_cpu",
247 "memory": "max_memory",
248 "nvidia.com/gpu": "max_gpu",
249}
252def check_resource_governance_fit(parsed: ParsedExample) -> list[StaticFinding]:
253 """Every example must be admissible under the default gco-jobs governance.
255 A container exceeding the LimitRange maxima is rejected at pod creation
256 with only namespace events explaining why, and the Job sits podless until
257 the caller gives up — the previous defaults rejected the platform's own
258 EFA training example exactly that way (live run ex241-df723811). Proving
259 the fit offline keeps the shipped examples and the shipped guardrails
260 from contradicting each other again. Only gco-jobs-namespaced pod specs
261 are checked: the LimitRange and ResourceQuota bind that namespace.
262 """
263 from gco.stacks.constants import DEFAULT_RESOURCE_QUOTA, parse_k8s_quantity
265 findings: list[StaticFinding] = []
266 for doc in parsed.documents:
267 kind = str(doc.get("kind", ""))
268 metadata = doc.get("metadata") or {}
269 if metadata.get("namespace", "gco-jobs") != "gco-jobs":
270 continue
271 pod_spec, replicas = _pod_spec_and_parallelism(doc, kind)
272 if pod_spec is None:
273 continue
274 containers = list(pod_spec.get("containers") or []) + list(
275 pod_spec.get("initContainers") or []
276 )
277 aggregate: dict[str, float] = dict.fromkeys(_QUOTA_DIMENSIONS, 0.0)
278 for container in containers:
279 resources = container.get("resources") or {}
280 requests = resources.get("requests") or {}
281 limits = resources.get("limits") or {}
282 for dimension, ceiling_key in _LIMIT_RANGE_DIMENSIONS.items():
283 ceiling = parse_k8s_quantity(DEFAULT_RESOURCE_QUOTA[ceiling_key])
284 for source_name, source in (("requests", requests), ("limits", limits)):
285 if dimension not in source:
286 continue
287 value = parse_k8s_quantity(source[dimension])
288 findings.append(
289 StaticFinding(
290 example=parsed.name,
291 check=(
292 f"LimitRange fit ({kind}/{metadata.get('name')}: "
293 f"{container.get('name')} {source_name}.{dimension})"
294 ),
295 passed=value <= ceiling,
296 detail=""
297 if value <= ceiling
298 else (
299 f"{source[dimension]} exceeds the default per-container "
300 f"ceiling {DEFAULT_RESOURCE_QUOTA[ceiling_key]} "
301 f"({ceiling_key})"
302 ),
303 )
304 )
305 for dimension in _QUOTA_DIMENSIONS:
306 if dimension in requests:
307 aggregate[dimension] += parse_k8s_quantity(requests[dimension])
308 for dimension, quota_key in _QUOTA_DIMENSIONS.items():
309 total = aggregate[dimension] * replicas
310 quota = parse_k8s_quantity(DEFAULT_RESOURCE_QUOTA[quota_key])
311 findings.append(
312 StaticFinding(
313 example=parsed.name,
314 check=(
315 f"ResourceQuota fit ({kind}/{metadata.get('name')}: "
316 f"{replicas}x pod requests.{dimension})"
317 ),
318 passed=total <= quota,
319 detail=""
320 if total <= quota
321 else (
322 f"aggregate {total:g} exceeds the default namespace quota "
323 f"{DEFAULT_RESOURCE_QUOTA[quota_key]} ({quota_key})"
324 ),
325 )
326 )
327 if parsed.spec.submission in {SUBMIT_API, SUBMIT_SQS}:
328 findings.extend(
329 _manifest_cap_findings(parsed, kind, str(metadata.get("name")), aggregate, replicas)
330 )
331 return findings
334#: Front-door budget dimensions (manifest/queue processor caps) by
335#: DEFAULT_MANIFEST_RESOURCE_CAPS key.
336_MANIFEST_CAP_DIMENSIONS = {
337 "cpu": "max_cpu_per_manifest",
338 "memory": "max_memory_per_manifest",
339 "nvidia.com/gpu": "max_gpu_per_manifest",
340}
343def _manifest_cap_findings(
344 parsed: ParsedExample,
345 kind: str,
346 name: str,
347 aggregate: dict[str, float],
348 replicas: int,
349) -> list[StaticFinding]:
350 """API/SQS-submitted manifests must also fit the front-door budget.
352 The manifest and queue processors cap what one submitted manifest may
353 total; an example the front door rejects while kubectl admits it (or
354 vice versa) means the layers contradict each other.
355 """
356 from gco.stacks.constants import DEFAULT_MANIFEST_RESOURCE_CAPS, parse_k8s_quantity
358 findings: list[StaticFinding] = []
359 for dimension, cap_key in _MANIFEST_CAP_DIMENSIONS.items():
360 total = aggregate[dimension] * replicas
361 cap = parse_k8s_quantity(DEFAULT_MANIFEST_RESOURCE_CAPS[cap_key])
362 findings.append(
363 StaticFinding(
364 example=parsed.name,
365 check=f"manifest-cap fit ({kind}/{name}: {replicas}x pod requests.{dimension})",
366 passed=total <= cap,
367 detail=""
368 if total <= cap
369 else (
370 f"aggregate {total:g} exceeds the default per-manifest cap "
371 f"{DEFAULT_MANIFEST_RESOURCE_CAPS[cap_key]} ({cap_key})"
372 ),
373 )
374 )
375 return findings
378def _pod_spec_and_parallelism(doc: dict[str, Any], kind: str) -> tuple[dict[str, Any] | None, int]:
379 """Extract the pod template spec and concurrent-pod count for a workload."""
380 spec = doc.get("spec") or {}
381 if kind == "Job":
382 template_spec = ((spec.get("template") or {}).get("spec")) or None
383 return template_spec, int(spec.get("parallelism", 1) or 1)
384 if kind in {"Deployment", "StatefulSet"}:
385 template_spec = ((spec.get("template") or {}).get("spec")) or None
386 return template_spec, int(spec.get("replicas", 1) or 1)
387 if kind == "Pod":
388 return spec or None, 1
389 if kind == "TrainJob":
390 # A TrainJob runs its spec.trainer view once per node; the shared
391 # decomposition builds the same synthetic pod spec the deployed
392 # validators check, so the offline governance math matches theirs.
393 from gco.services.manifest_processor import extract_trainjob_pod_specs
395 trainjob_specs = extract_trainjob_pod_specs(doc)
396 return trainjob_specs.trainer, trainjob_specs.num_nodes
397 return None, 1
400def run_static_checks(repo_root: Path, names: list[str] | None = None) -> list[StaticFinding]:
401 """Run every offline check; returns findings (all must pass)."""
402 findings = check_registry_symmetry(repo_root)
403 for name in names or example_names(repo_root):
404 findings.append(check_spec_shape(name))
405 if EXAMPLE_SPECS.get(name) is None:
406 continue
407 parsed = parse_example(repo_root, name)
408 findings.append(check_submission_matches_catalog(repo_root, name))
409 findings.extend(check_transport_acceptance(parsed))
410 findings.extend(check_namespaces(parsed))
411 findings.extend(check_resource_governance_fit(parsed))
412 return findings