Coverage for .github / scripts / validate_k8s_manifests.py: 100.00%
287 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Validate Kubernetes manifests with kubeconform (schema, not just YAML syntax).
3Confirms that every hand-authored manifest this repo ships — the
4kubectl-applier's own manifests and the `examples/` gallery — is not just
5parseable YAML but a *schema-valid* Kubernetes (or supported CRD) resource.
6`kubeconform` (https://github.com/yannh/kubeconform) does the actual schema
7check; this script exists to bridge three gaps kubeconform can't close on its
8own:
10 Template placeholders: `lambda/kubectl-applier-simple/manifests/*.yaml`
11 contains `{{PLACEHOLDER}}` tokens the kubectl-applier Lambda substitutes
12 at deploy time (see `handler.py` / `regional_stack.py`). Raw, these
13 aren't valid YAML in several spots (bare scalars, a block-list
14 placeholder), so kubeconform can't even parse the file. This script
15 renders every placeholder to a schema-shaped stub first — enough to make
16 the YAML parse and the field types check out, not a claim about the real
17 runtime value.
19 Non-Kubernetes files under `examples/`: that directory also ships a DAG
20 orchestration file (`pipeline-dag.yaml`, a GCO-specific format, not a K8s
21 manifest) and JSON metric/state fixtures used by other examples. Those
22 are excluded by construction (only `*.yaml`/`*.yml` under `examples/` are
23 scanned, and `pipeline-dag.yaml` is skipped by name).
25 Defaulted required fields: a CRD can mark a field required and also give
26 it a default. The API server fills the default in before it validates,
27 so a manifest may leave the field out, but kubeconform validates the
28 document as written and reports the field missing. For the fields in
29 `_API_SERVER_DEFAULTS` (kro's `ResourceGraphDefinition`
30 `spec.schema.scope`), the rendered copy gets the value the API server
31 would fill in.
33Schema resolution: kubeconform's bundled catalog only covers upstream
34Kubernetes, and is resolved at the cluster's minor (`-kubernetes-version
35<cdk.json kubernetes_version>.0`, see `kubernetes_schema_version()`) rather
36than kubeconform's floating `master` default, so a field the deployed API
37server does not know yet fails here. CRDs this repo depends on (Karpenter `NodePool`, the AWS Load
38Balancer Controller Gateway API configuration CRDs, Kueue, KEDA) are resolved
39via the community datreeio/CRDs-catalog as a second `-schema-location`. Two CRDs used in
40`examples/` aren't in that catalog yet (KubeRay's `RayCluster`, Volcano's
41`Job`) — those are validated for YAML-shape only (via the structural check)
42and explicitly `-skip`ped in the schema pass so kubeconform doesn't report a
43false "no schema found" error for them.
45Usage::
47 # Validate both directories with the real kubeconform binary (the CI gate):
48 python3 .github/scripts/validate_k8s_manifests.py
50 # Validate a specific directory, file, or quoted glob. --path is repeatable:
51 python3 .github/scripts/validate_k8s_manifests.py --path examples/simple-job.yaml
52 python3 .github/scripts/validate_k8s_manifests.py --path 'examples/**/*.yaml'
54 # Point at a different kubeconform binary:
55 python3 .github/scripts/validate_k8s_manifests.py --kubeconform-binary /usr/local/bin/kubeconform
57Exit codes::
59 0 every scanned manifest is schema-valid (or intentionally skipped)
60 1 one or more manifests failed validation
61 2 unexpected I/O / argument error (kubeconform missing, directory absent)
63The module is importable from the test suite — call ``render_placeholders()``,
64``apply_api_server_defaults()``, ``collect_target_files()``, or
65``iter_target_files()`` directly to exercise the logic without invoking the
66binary.
67"""
69from __future__ import annotations
71import argparse
72import glob
73import json
74import re
75import shutil
76import subprocess # used only to invoke the pinned `kubeconform` binary with fixed argv
77import sys
78import tempfile
79from pathlib import Path
80from typing import Any
82import yaml
84_REPO_ROOT = Path(__file__).resolve().parents[2]
86# Directories this script scans by default. Each entry is a directory
87# relative to the repo root; every ``*.yaml``/``*.yml`` file directly inside
88# it is a candidate (no recursion: nested directories are listed explicitly).
89DEFAULT_TARGET_DIRS = (
90 "lambda/kubectl-applier-simple/manifests",
91 "examples",
92 # The Git path examples/argocd-gitops-job.yaml syncs: never submitted
93 # through GCO, so this schema check is its only pre-cluster validation.
94 "examples/gitops/hello-job",
95)
97# Files that are YAML but not Kubernetes manifests, so they're excluded by
98# name rather than left for kubeconform to fail on a missing 'kind'/'apiVersion'.
99# - pipeline-dag.yaml: a GCO DAG-orchestration definition (see `gco dag`,
100# tests/test_mission_validation.py-style docs). Its own structure is
101# checked separately in CI (integration:k8s:manifest-schema's "Validate
102# DAG pipeline definitions" step) — validating it against the K8s schema
103# would always fail because it deliberately has no 'kind'.
104NON_MANIFEST_FILENAMES = frozenset({"pipeline-dag.yaml"})
106# GVK-qualified (not bare-Kind) skips for CRDs used in examples/ that the
107# datreeio/CRDs-catalog fallback doesn't (yet) carry a schema for. GVK
108# qualification matters here specifically because Volcano's Job kind
109# collides with the built-in batch/v1 Job — a bare `-skip Job` would also
110# skip every ordinary Job manifest in the repo.
111SCHEMA_UNAVAILABLE_SKIPS = (
112 "ray.io/v1/RayCluster", # KubeRay — not in datreeio/CRDs-catalog
113 "batch.volcano.sh/v1alpha1/Job", # Volcano — not in datreeio/CRDs-catalog
114 # Kubeflow Trainer v2 — not in datreeio/CRDs-catalog. The shipped
115 # torch-distributed runtime is validated more strongly than a catalog
116 # lookup could: validate_helm_charts.py's trainer runtime lockstep
117 # re-renders the pinned kubeflow-trainer chart online and requires the
118 # manifest to reproduce it spec-for-spec; the TrainJob example's shape
119 # is exercised by the submission pipeline's decomposition tests and the
120 # live example run.
121 "trainer.kubeflow.org/v1alpha1/TrainJob",
122 "trainer.kubeflow.org/v1alpha1/ClusterTrainingRuntime",
123 # AWS LBC gateway CRDs at their v1 storage version (v3.5.0+) — the
124 # datreeio catalog only carries the deprecated v1beta1 schemas. These two
125 # resources are instead schema-validated against the exact pinned CRD
126 # bundle by validate_helm_charts.py's gateway-lockstep check, which is
127 # stronger than the catalog lookup this skip bypasses.
128 "gateway.k8s.aws/v1/LoadBalancerConfiguration",
129 "gateway.k8s.aws/v1/TargetGroupConfiguration",
130 # Slinky slurm-operator Token (post-helm-slurm-api-access.yaml) — the
131 # catalog carries no slinky.slurm.net schemas. The manifest's spec keys
132 # are pinned against the v1.2.3 CRD by tests/test_kubectl_applier.py
133 # (TestSlurmApiAccessManifest), and the live Slurm validation action
134 # applies it to a real API server serving that CRD.
135 "slinky.slurm.net/v1beta1/Token",
136 # The example-defined BatchJob APIs. No catalog can carry them: their
137 # schemas are the companion examples themselves (the Crossplane XRD in
138 # crossplane-batch-api.yaml, the kro RGD in kro-batch-api.yaml), which ARE
139 # schema-validated here, and the example harness applies each instance to a
140 # real API server serving that schema.
141 "examples.gco.io/v1alpha1/BatchJob",
142 "kro.run/v1alpha1/BatchJob",
143)
145# Required fields a CRD also defaults, keyed by (apiVersion, kind): the path to
146# each field and the value the API server fills in. A structural schema's
147# defaults are applied before the object is validated, so a manifest may leave
148# such a field out, but kubeconform validates the document as written and would
149# report it missing. apply_api_server_defaults() adds the default to the
150# rendered copy only. A value the manifest sets is kept, and a missing parent
151# object is left for kubeconform to report.
152#
153# kro.run/v1alpha1 ResourceGraphDefinition spec.schema.scope: kro v0.9.0
154# added it (Namespaced or Cluster, default Namespaced, immutable once set),
155# and the datreeio catalog has marked it required since its kro v0.9.4
156# refresh (2026-10-18). examples/kro-batch-api.yaml leaves it out on
157# purpose: a kro older than v0.9.0 does not know the field, and kubectl's
158# strict field validation would then reject the whole definition.
159_API_SERVER_DEFAULTS: dict[tuple[str, str], tuple[tuple[tuple[str, ...], str], ...]] = {
160 ("kro.run/v1alpha1", "ResourceGraphDefinition"): ((("spec", "schema", "scope"), "Namespaced"),),
161}
163# kubeconform's own default schema catalog (upstream Kubernetes resources).
164DEFAULT_SCHEMA_LOCATION = "default"
166# The upstream schemas are validated at the cluster's minor, not kubeconform's
167# floating ``master`` default: ``-kubernetes-version`` is derived from
168# ``context.kubernetes_version`` in cdk.json (the single source for the EKS
169# minor) with patch ``0``. Built-in resource schemas do not change within a
170# minor, every ``vX.Y.0`` is in the catalog kubeconform fetches from
171# (yannh/kubernetes-json-schema) long before EKS ships that minor, and no
172# patch bump elsewhere in the repo can race the catalog. So a manifest that
173# uses a field the cluster's API server does not yet know fails here instead
174# of at deploy time, and nothing changes in this script when the minor moves.
175CDK_JSON_PATH = _REPO_ROOT / "cdk.json"
176_KUBERNETES_MINOR_RE = re.compile(r"^(\d+)\.(\d+)$")
179def kubernetes_schema_version(cdk_json: Path = CDK_JSON_PATH) -> str:
180 """Return the ``X.Y.0`` kubeconform ``-kubernetes-version`` for the cluster minor.
182 Raises ``ValueError`` when cdk.json has no usable
183 ``context.kubernetes_version`` (the gate must not silently fall back to
184 ``master``).
185 """
186 try:
187 context = json.loads(cdk_json.read_text(encoding="utf-8")).get("context", {})
188 except (OSError, json.JSONDecodeError) as exc:
189 raise ValueError(f"cannot read {cdk_json}: {exc}") from exc
190 minor = context.get("kubernetes_version")
191 if not isinstance(minor, str) or not _KUBERNETES_MINOR_RE.fullmatch(minor.strip()):
192 raise ValueError(
193 f"{cdk_json}: context.kubernetes_version must be a 'major.minor' string "
194 f"(for example '1.37'), got {minor!r}"
195 )
196 return f"{minor.strip()}.0"
199# Community-maintained CRD catalog covering Karpenter, EKS Auto Mode, Kueue,
200# KEDA, and hundreds of other CRDs — see https://github.com/datreeio/CRDs-catalog.
201# kubeconform tries -schema-location entries in order and stops at the first
202# match, so this is consulted only when a Kind isn't a built-in K8s resource.
203CRD_CATALOG_SCHEMA_LOCATION = (
204 "https://raw.githubusercontent.com/datreeio/CRDs-catalog/main/"
205 "{{.Group}}/{{.ResourceKind}}_{{.ResourceAPIVersion}}.json"
206)
208_PLACEHOLDER_RE = re.compile(r"\{\{[A-Za-z_]+\}\}")
210# A handful of placeholders sit in *structural* or *typed* positions where a
211# generic string stub would either break YAML parsing or fail the schema's
212# type check. Each needs a stub shaped like the real substitution.
213#
214# {{VPC_ENDPOINT_CIDR_BLOCKS}} expands to one or more `- ipBlock: {cidr: ...}`
215# list entries (see regional_stack.py::_compute... / the NetworkPolicy
216# `to:` block in 03-network-policies.yaml) — the placeholder sits at the
217# start of a YAML sequence, so it needs a real sequence item, not a string.
218#
219# {{ARGOCD_SOURCE_REPOS}} / {{ARGOCD_GITOPS_SYNC_POLICY}} are rendered by
220# gco/argocd_config.py::compute_argocd_replacements as single-line JSON (a
221# YAML flow sequence / mapping) — the AppProject ``sourceRepos`` list in
222# post-helm-argocd-access.yaml and the root Application ``syncPolicy`` object
223# in post-helm-argocd-gitops.yaml — so their stubs must be JSON of the same
224# shape, not a string.
225_STRUCTURAL_STUBS: dict[str, str] = {
226 "{{VPC_ENDPOINT_CIDR_BLOCKS}}": '- ipBlock:\n cidr: "10.0.0.0/16"',
227 "{{ARGOCD_SOURCE_REPOS}}": '["https://github.com/example/*"]',
228 "{{ARGOCD_GITOPS_SYNC_POLICY}}": '{"automated": {"selfHeal": true, "prune": false}}',
229}
231# Placeholders that sit in a bare (unquoted) numeric scalar position — e.g.
232# `pollingInterval: {{QP_POLLING_INTERVAL}}` in post-helm-sqs-consumer.yaml.
233# These must render to a bare integer, not a quoted string, or the field
234# fails the schema's `type: integer` check.
235_INTEGER_PLACEHOLDER_TOKENS: frozenset[str] = frozenset(
236 {
237 "{{INFERENCE_PROXY_TLS_CPU_TARGET_UTILIZATION}}",
238 "{{INFERENCE_PROXY_MIN_REPLICAS}}",
239 "{{INFERENCE_PROXY_MAX_REPLICAS}}",
240 "{{MP_REPLICAS}}",
241 "{{MP_HPA_MAX_REPLICAS}}",
242 "{{MP_HPA_CPU_TARGET_UTILIZATION}}",
243 "{{QP_POLLING_INTERVAL}}",
244 "{{QP_SUCCESSFUL_JOBS_HISTORY}}",
245 "{{QP_FAILED_JOBS_HISTORY}}",
246 "{{QP_MAX_CONCURRENT_JOBS}}",
247 }
248)
250# Placeholders that sit in Kubernetes *quantity* positions (e.g. the Kueue
251# ClusterQueue `nominalQuota` fields in post-helm-kueue-default-queues.yaml,
252# which reuse the namespace-quota tokens). CRD catalog schemas enforce the
253# quantity regex, which the generic stub fails; a literal 1 is valid whether
254# the position is quoted or bare.
255_QUANTITY_PLACEHOLDER_TOKENS: frozenset[str] = frozenset(
256 {
257 "{{INFERENCE_PROXY_TLS_CPU_REQUEST}}",
258 "{{MP_CPU_LIMIT}}",
259 "{{MP_MEMORY_LIMIT}}",
260 "{{QUOTA_MAX_CPU}}",
261 "{{QUOTA_MAX_MEMORY}}",
262 "{{QUOTA_MAX_GPU}}",
263 # The Slurm accounting database's volumeClaimTemplate size
264 # (29-slurm-accounting-db.yaml).
265 "{{SLURM_ACCOUNTING_DB_STORAGE_SIZE}}",
266 }
267)
269# Every other placeholder (quoted string values, and the couple of bare
270# `image: {{...}}` lines) renders fine as a generic string stub — YAML
271# treats an unquoted bare word as a string scalar automatically.
272_GENERIC_STUB = "placeholder-value"
275def render_placeholders(text: str) -> str:
276 """Replace every ``{{PLACEHOLDER}}`` token with a schema-shaped stub.
278 This is a validation-time rendering only — it exists to make templated
279 manifests parseable and schema-checkable, not to model what the
280 kubectl-applier Lambda actually substitutes at deploy time. See the
281 module docstring for why each stub category exists.
282 """
283 for token, stub in _STRUCTURAL_STUBS.items():
284 text = text.replace(token, stub)
285 for token in _INTEGER_PLACEHOLDER_TOKENS | _QUANTITY_PLACEHOLDER_TOKENS:
286 text = text.replace(token, "1")
287 return _PLACEHOLDER_RE.sub(_GENERIC_STUB, text)
290def apply_api_server_defaults(text: str) -> str:
291 """Add the defaults the API server would fill in (see ``_API_SERVER_DEFAULTS``).
293 Text in which no document needs a default is returned unchanged, so every
294 other file reaches kubeconform exactly as rendered. Text that is not valid
295 YAML is also returned unchanged, for kubeconform to report.
296 """
297 if not any(kind in text for _api_version, kind in _API_SERVER_DEFAULTS):
298 return text
299 try:
300 documents = list(yaml.safe_load_all(text))
301 except yaml.YAMLError:
302 return text
303 changed = False
304 for document in documents:
305 if not isinstance(document, dict):
306 continue
307 api_version, kind = document.get("apiVersion"), document.get("kind")
308 if not isinstance(api_version, str) or not isinstance(kind, str):
309 continue
310 for path, value in _API_SERVER_DEFAULTS.get((api_version, kind), ()):
311 parent: Any = document
312 for key in path[:-1]:
313 parent = parent.get(key) if isinstance(parent, dict) else None
314 if isinstance(parent, dict) and path[-1] not in parent:
315 parent[path[-1]] = value
316 changed = True
317 if not changed:
318 return text
319 dumped: str = yaml.safe_dump_all(
320 [document for document in documents if document is not None], sort_keys=False
321 )
322 return dumped
325def _manifest_files_in_directory(directory: Path) -> list[Path]:
326 """Return supported direct-child manifests from one directory."""
327 files: list[Path] = []
328 for pattern in ("*.yaml", "*.yml"):
329 for path in sorted(directory.glob(pattern)):
330 if path.name not in NON_MANIFEST_FILENAMES:
331 files.append(path.resolve())
332 return files
335def collect_target_files(targets: tuple[str, ...]) -> tuple[list[Path], list[str]]:
336 """Resolve every explicit directory, file, or glob independently.
338 Valid files are returned even when another input is invalid so callers can
339 still validate and report them. Errors preserve one entry per bad explicit
340 input. Directories retain the historical direct-child-only behavior; quoted
341 globs can opt into recursion with ``**``.
342 """
343 files: list[Path] = []
344 errors: list[str] = []
345 seen: set[Path] = set()
347 def add(path: Path) -> None:
348 resolved = path.resolve()
349 if resolved not in seen:
350 seen.add(resolved)
351 files.append(resolved)
353 for raw_target in targets:
354 expanded = Path(raw_target).expanduser()
355 resolved_input = expanded if expanded.is_absolute() else _REPO_ROOT / expanded
357 if glob.has_magic(str(expanded)):
358 matches = [
359 Path(match) for match in sorted(glob.glob(str(resolved_input), recursive=True))
360 ]
361 matched_manifests: list[Path] = []
362 for match in matches:
363 if match.is_dir():
364 matched_manifests.extend(_manifest_files_in_directory(match))
365 elif (
366 match.is_file()
367 and match.suffix in (".yaml", ".yml")
368 and match.name not in NON_MANIFEST_FILENAMES
369 ):
370 matched_manifests.append(match.resolve())
371 if not matched_manifests:
372 errors.append(f"{raw_target}: glob matched no Kubernetes YAML manifests")
373 continue
374 for path in matched_manifests:
375 add(path)
376 continue
378 if not resolved_input.exists():
379 errors.append(f"{raw_target}: path does not exist")
380 continue
381 if resolved_input.is_dir():
382 directory_files = _manifest_files_in_directory(resolved_input)
383 if not directory_files:
384 errors.append(f"{raw_target}: directory contains no Kubernetes YAML manifests")
385 continue
386 for path in directory_files:
387 add(path)
388 continue
389 if resolved_input.is_file():
390 if resolved_input.suffix not in (".yaml", ".yml"):
391 errors.append(f"{raw_target}: explicit file is not .yaml or .yml")
392 elif resolved_input.name in NON_MANIFEST_FILENAMES:
393 errors.append(f"{raw_target}: explicit file is not a Kubernetes manifest")
394 else:
395 add(resolved_input)
396 continue
397 errors.append(f"{raw_target}: unsupported input type")
399 return files, errors
402def iter_target_files(target_dirs: tuple[str, ...] = DEFAULT_TARGET_DIRS) -> list[Path]:
403 """Compatibility wrapper returning valid manifest files only.
405 Use :func:`collect_target_files` when input errors must be surfaced.
406 """
407 files, _errors = collect_target_files(target_dirs)
408 return files
411def _rendered_relative_path(path: Path) -> Path:
412 """Return a collision-safe relative location for a rendered source file."""
413 resolved = path.resolve()
414 try:
415 return resolved.relative_to(_REPO_ROOT.resolve())
416 except ValueError:
417 # Explicit absolute paths outside the repository are still supported.
418 # Preserve their path below a marker rather than flattening basenames.
419 return Path("_external", *resolved.parts[1:])
422def render_tree(files: list[Path], dest: Path) -> list[Path]:
423 """Render files beneath ``dest`` while preserving source-relative paths.
425 Repository files retain their repository-relative path, preventing files
426 with the same basename in different inputs from overwriting one another.
427 Returns the rendered paths for callers/tests that need the mapping.
428 """
429 rendered_paths: list[Path] = []
430 dest.mkdir(parents=True, exist_ok=True)
431 for path in files:
432 text = path.read_text(encoding="utf-8")
433 rendered = render_placeholders(text) if "{{" in text else text
434 rendered = apply_api_server_defaults(rendered)
435 rendered_path = dest / _rendered_relative_path(path)
436 rendered_path.parent.mkdir(parents=True, exist_ok=True)
437 rendered_path.write_text(rendered, encoding="utf-8")
438 rendered_paths.append(rendered_path)
439 return rendered_paths
442def run_kubeconform(
443 directory: Path,
444 *,
445 kubeconform_binary: str = "kubeconform",
446 strict: bool = True,
447 extra_schema_locations: tuple[str, ...] = (CRD_CATALOG_SCHEMA_LOCATION,),
448 skip_gvks: tuple[str, ...] = SCHEMA_UNAVAILABLE_SKIPS,
449 kubernetes_version: str | None = None,
450) -> tuple[int, object]:
451 """Run kubeconform against every manifest in ``directory``, JSON output.
453 ``kubernetes_version`` is the ``-kubernetes-version`` the upstream schemas
454 are resolved at; ``None`` derives it from cdk.json via
455 :func:`kubernetes_schema_version`.
457 Returns ``(returncode, parsed_json)``. ``parsed_json`` is ``{}`` if
458 kubeconform produced no parseable JSON (e.g. the binary is missing —
459 callers should check that separately via ``shutil.which`` before calling).
460 """
461 if kubernetes_version is None:
462 kubernetes_version = kubernetes_schema_version()
463 cmd = [
464 kubeconform_binary,
465 "-output",
466 "json",
467 "-summary",
468 "-verbose",
469 "-kubernetes-version",
470 kubernetes_version,
471 "-schema-location",
472 DEFAULT_SCHEMA_LOCATION,
473 ]
474 for location in extra_schema_locations:
475 cmd.extend(["-schema-location", location])
476 if skip_gvks:
477 cmd.extend(["-skip", ",".join(skip_gvks)])
478 if strict:
479 cmd.append("-strict")
480 cmd.append(str(directory))
482 proc = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed argv, no shell=True
483 cmd,
484 capture_output=True,
485 text=True,
486 check=False,
487 timeout=180,
488 )
489 try:
490 parsed = json.loads(proc.stdout) if proc.stdout.strip() else {}
491 except json.JSONDecodeError:
492 parsed = {}
493 return proc.returncode, parsed
496_KUBECONFORM_STATUS_TO_SUMMARY = {
497 "statusValid": "valid",
498 "statusInvalid": "invalid",
499 "statusError": "errors",
500 "statusSkipped": "skipped",
501}
502_KUBECONFORM_RESOURCE_FIELDS = frozenset({"filename", "kind", "name", "version", "status", "msg"})
503_KUBECONFORM_BLANK_RESOURCE_FIELDS = ("kind", "name", "version", "status", "msg")
506def _is_kubeconform_separator_record(resource: object) -> bool:
507 """Recognize kubeconform v0.8.0's record for a leading YAML separator.
509 The pinned binary emits one record with a filename but empty semantic
510 fields when a file starts with ``---``. Match that exact shape so a
511 partially populated record or a future unknown field still fails closed.
512 """
513 return (
514 isinstance(resource, dict)
515 and set(resource) == _KUBECONFORM_RESOURCE_FIELDS
516 and isinstance(resource["filename"], str)
517 and bool(resource["filename"])
518 and all(resource[field] == "" for field in _KUBECONFORM_BLANK_RESOURCE_FIELDS)
519 )
522def validate_kubeconform_output(
523 result: object,
524 *,
525 expected_filenames: set[str],
526) -> list[str]:
527 """Validate kubeconform's JSON envelope and per-file resource accounting.
529 A zero process exit code is not sufficient: blank, malformed, truncated,
530 or structurally incomplete JSON must fail closed rather than reporting
531 ``OK: 0 manifest(s)``. Multi-document inputs may produce many resource
532 records, so completeness is based on exact rendered filenames instead of
533 comparing aggregate record and file counts.
534 """
535 if not isinstance(result, dict):
536 return ["top-level JSON value is not an object"]
538 errors: list[str] = []
539 raw_resources = result.get("resources")
540 if not isinstance(raw_resources, list):
541 return ["'resources' is missing or is not an array"]
542 resources = [
543 resource for resource in raw_resources if not _is_kubeconform_separator_record(resource)
544 ]
545 if not resources:
546 errors.append("'resources' is empty")
548 expected = {str(Path(filename).resolve(strict=False)) for filename in expected_filenames}
549 observed: set[str] = set()
550 actual_counts = dict.fromkeys(_KUBECONFORM_STATUS_TO_SUMMARY.values(), 0)
551 for index, resource in enumerate(resources):
552 if not isinstance(resource, dict):
553 errors.append(f"resources[{index}] is not an object")
554 continue
556 filename = resource.get("filename")
557 if not isinstance(filename, str) or not filename:
558 errors.append(f"resources[{index}].filename is missing or is not a non-empty string")
559 else:
560 observed.add(str(Path(filename).resolve(strict=False)))
562 status = resource.get("status")
563 summary_field = (
564 _KUBECONFORM_STATUS_TO_SUMMARY.get(status) if isinstance(status, str) else None
565 )
566 if summary_field is None:
567 errors.append(f"resources[{index}] has unknown status {status!r}")
568 continue
569 actual_counts[summary_field] += 1
571 missing = sorted(expected - observed)
572 if missing:
573 errors.append("no resource result was returned for input file(s): " + ", ".join(missing))
574 unexpected = sorted(observed - expected)
575 if unexpected:
576 errors.append(
577 "resource results were returned for unexpected file(s): " + ", ".join(unexpected)
578 )
580 summary = result.get("summary")
581 if not isinstance(summary, dict):
582 errors.append("'summary' is missing or is not an object")
583 return errors
585 for field, actual_count in actual_counts.items():
586 reported_count = summary.get(field)
587 if (
588 isinstance(reported_count, bool)
589 or not isinstance(reported_count, int)
590 or reported_count < 0
591 ):
592 errors.append(f"summary.{field} is missing or is not a non-negative integer")
593 elif reported_count != actual_count:
594 errors.append(
595 f"summary.{field} reports {reported_count}, but resources contain {actual_count}"
596 )
598 return errors
601def format_failures(result: dict[str, Any]) -> list[str]:
602 """Turn kubeconform's per-resource JSON records into readable error lines.
604 Only ``statusInvalid`` and ``statusError`` are failures — ``statusValid``
605 and ``statusSkipped`` (the two explicitly ``-skip``ped CRDs) are fine.
606 """
607 lines: list[str] = []
608 for resource in result.get("resources", []):
609 status = resource.get("status")
610 if status not in ("statusInvalid", "statusError"):
611 continue
612 filename = resource.get("filename", "<unknown file>")
613 kind = resource.get("kind") or "<unparsed>"
614 name = resource.get("name") or ""
615 label = f"{kind} {name}".strip()
616 msg = resource.get("msg", "")
617 lines.append(f"{filename}: {label}: {msg}")
618 return lines
621def _build_parser() -> argparse.ArgumentParser:
622 parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
623 parser.add_argument(
624 "--path",
625 action="append",
626 dest="paths",
627 help=(
628 "Directory, YAML file, or glob to validate (relative to repo root or absolute). "
629 "Repeatable; quote glob patterns so Python expands them. Defaults to both "
630 "lambda/kubectl-applier-simple/manifests and examples."
631 ),
632 )
633 parser.add_argument(
634 "--kubeconform-binary",
635 default="kubeconform",
636 help="kubeconform executable to use (default: 'kubeconform' on PATH).",
637 )
638 parser.add_argument(
639 "--no-strict",
640 action="store_true",
641 help="Disable kubeconform's -strict mode (allow unknown/additional properties).",
642 )
643 parser.add_argument(
644 "-v",
645 "--verbose",
646 action="store_true",
647 help="Print a per-resource OK line in addition to failures.",
648 )
649 return parser
652def _print_input_errors(errors: list[str]) -> None:
653 if not errors:
654 return
655 print(f"ERROR: {len(errors)} manifest input problem(s) found:", file=sys.stderr)
656 for error in errors:
657 print(f" - {error}", file=sys.stderr)
660def main(argv: list[str] | None = None) -> int:
661 args = _build_parser().parse_args(argv)
662 targets = tuple(args.paths) if args.paths else DEFAULT_TARGET_DIRS
663 files, input_errors = collect_target_files(targets)
665 if shutil.which(args.kubeconform_binary) is None:
666 _print_input_errors(input_errors)
667 print(
668 f"ERROR: '{args.kubeconform_binary}' not found on PATH. "
669 "Install it (see Dockerfile.dev / docs/MAINTENANCE.md) or pass "
670 "--kubeconform-binary.",
671 file=sys.stderr,
672 )
673 return 2
675 if not files:
676 _print_input_errors(input_errors)
677 if not input_errors:
678 print(f"ERROR: no *.yaml/*.yml files found for {targets}", file=sys.stderr)
679 return 2
681 try:
682 kubernetes_version = kubernetes_schema_version()
683 except ValueError as exc:
684 _print_input_errors(input_errors)
685 print(f"ERROR: cannot derive the Kubernetes schema version: {exc}", file=sys.stderr)
686 return 2
688 with tempfile.TemporaryDirectory(prefix="gco-k8s-validate-") as tmp:
689 rendered_dir = Path(tmp)
690 rendered_paths = render_tree(files, rendered_dir)
691 expected_filenames = {str(path) for path in rendered_paths}
693 rc, result = run_kubeconform(
694 rendered_dir,
695 kubeconform_binary=args.kubeconform_binary,
696 strict=not args.no_strict,
697 kubernetes_version=kubernetes_version,
698 )
700 output_errors = validate_kubeconform_output(
701 result,
702 expected_filenames=expected_filenames,
703 )
704 result_dict = result if isinstance(result, dict) else {}
706 if args.verbose and not output_errors:
707 for resource in result_dict.get("resources", []):
708 if resource.get("status") == "statusValid":
709 kind = resource.get("kind") or ""
710 name = resource.get("name") or ""
711 print(f"ok {resource.get('filename')}: {kind} {name}".rstrip())
713 failures = format_failures(result_dict) if not output_errors else []
714 summary = result_dict.get("summary", {})
716 if failures:
717 print()
718 print(f"ERROR: {len(failures)} Kubernetes manifest validation problem(s) found:")
719 for line in failures:
720 print(f" - {line}")
721 print()
722 print(
723 "Fix the manifest (or, if this is a new CRD with no upstream schema "
724 "yet, add it to SCHEMA_UNAVAILABLE_SKIPS in "
725 "validate_k8s_manifests.py with a comment explaining why; if the "
726 "catalog marks a field required that the API server defaults, add "
727 "it to _API_SERVER_DEFAULTS instead)."
728 )
730 _print_input_errors(input_errors)
732 runtime_failure = bool(output_errors) or (rc != 0 and not failures)
733 if output_errors:
734 print("ERROR: kubeconform returned unusable JSON output:", file=sys.stderr)
735 for error in output_errors:
736 print(f" - {error}", file=sys.stderr)
737 elif rc != 0 and not failures:
738 # A non-zero process result without resource validation failures is an
739 # invocation/runtime error, even if a partial JSON document was emitted.
740 print("ERROR: kubeconform exited non-zero without validation failures.", file=sys.stderr)
742 # Input/runtime errors take precedence, but valid supplied files were still
743 # rendered and validated above so one missing path cannot mask their report.
744 if input_errors or runtime_failure:
745 return 2
746 if failures:
747 return 1
749 print(
750 f"OK: {summary.get('valid', 0)} manifest(s) are schema-valid against Kubernetes "
751 f"{kubernetes_version} ({summary.get('skipped', 0)} intentionally skipped: GVKs "
752 "with no catalog schema, listed in SCHEMA_UNAVAILABLE_SKIPS)."
753 )
754 return 0
757if __name__ == "__main__":
758 sys.exit(main())