Coverage for scripts / example_job_validation / drivers.py: 100.00%
638 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Submission, success-criteria, setup, and cleanup drivers for one example.
3Every driver takes the parsed example plus the run plumbing and returns
4evidence dictionaries for the report. Submission always travels the
5DOCUMENTED path (the real ``gco`` CLI or ``kubectl apply``); any deliberate
6manifest mutation (spec.mutations) is applied to a disclosed temp copy.
7"""
9from __future__ import annotations
11import json
12import re
13import subprocess
14import tempfile
15import threading
16import time
17from dataclasses import dataclass, field
18from pathlib import Path
19from typing import Any
21import yaml
23from .kube import KubectlRunner, session_environment, through_tunnel
24from .specs import (
25 ACK_RESOURCE_SYNCED,
26 ARGOCD_APP_HEALTHY,
27 COMPOSED_JOB_COMPLETES,
28 DAG_RUN,
29 DEPLOYMENT_AVAILABLE,
30 JOB_COMPLETES,
31 KUBECTL_APPLY,
32 RAYCLUSTER_READY,
33 SCALEDJOB_SCALES,
34 SUBMIT_API,
35 SUBMIT_DIRECT,
36 SUBMIT_SQS,
37 TRAINJOB_COMPLETES,
38 VCJOB_COMPLETES,
39)
40from .static_checks import ParsedExample, application_source_documents
42#: boto3 Session.client() is not thread-safe (client creation mutates shared
43#: loader state); every client creation against the run's shared session must
44#: hold this lock when examples run in parallel. The created clients ARE safe
45#: to use concurrently.
46BOTO_CLIENT_LOCK = threading.Lock()
48_POLL_SECONDS = 15
51class ExampleValidationError(RuntimeError):
52 """One example failed its criteria; the message carries the evidence."""
55@dataclass
56class ExampleRunResult:
57 """Evidence for one example's live validation."""
59 name: str
60 status: str # passed | failed | skipped
61 submission: str
62 duration_seconds: float = 0.0
63 detail: str = ""
64 mutations: dict[str, str] = field(default_factory=dict)
65 evidence: dict[str, Any] = field(default_factory=dict)
67 def to_dict(self) -> dict[str, Any]:
68 return {
69 "name": self.name,
70 "status": self.status,
71 "submission": self.submission,
72 "duration_seconds": round(self.duration_seconds, 3),
73 "detail": self.detail,
74 "mutations": self.mutations,
75 "evidence": self.evidence,
76 }
79def _run_cli(
80 args: list[str],
81 repo_root: Path,
82 timeout: int = 600,
83 env: dict[str, str] | None = None,
84) -> tuple[int, str, str]:
85 result = subprocess.run(
86 args, cwd=repo_root, capture_output=True, text=True, timeout=timeout, check=False, env=env
87 )
88 return result.returncode, result.stdout, result.stderr
91def write_temp_manifest(documents: list[dict[str, Any]], suffix: str) -> Path:
92 """Write documents to a private temp file and return its path."""
93 fd, name = tempfile.mkstemp(suffix=suffix, text=True)
94 with open(fd, "w", encoding="utf-8") as fh:
95 yaml.safe_dump_all(documents, fh)
96 return Path(name)
99def apply_mutations(parsed: ParsedExample) -> tuple[Path, dict[str, str]]:
100 """Materialize the manifest to submit: verbatim, or a disclosed mutated copy.
102 Mutation keys use the shape ``Deployment.env.NAME`` (replace that env
103 var's value) or ``Deployment.args.--flag`` (replace the argv element
104 following the flag). Anything else is a spec bug and raises.
105 """
106 if not parsed.spec.mutations:
107 return parsed.path, {}
108 from .specs import REMOVE_VALUE
110 documents = [dict(doc) for doc in parsed.documents]
111 for key, replacement in parsed.spec.mutations.items():
112 kind, channel, target = key.split(".", 2)
113 for doc in documents:
114 if doc.get("kind") != kind:
115 continue
116 containers = doc["spec"]["template"]["spec"]["containers"]
117 for container in containers:
118 if channel == "env":
119 env_entries = container.get("env", [])
120 if replacement == REMOVE_VALUE:
121 container["env"] = [
122 entry for entry in env_entries if entry.get("name") != target
123 ]
124 continue
125 for env_entry in env_entries:
126 if env_entry.get("name") == target:
127 env_entry["value"] = replacement
128 elif channel == "args":
129 args = container.get("args", [])
130 for index, value in enumerate(args):
131 if value == target and index + 1 < len(args):
132 args[index + 1] = replacement
133 else:
134 raise ValueError(f"Unsupported mutation channel in {key!r}")
135 return (
136 write_temp_manifest(documents, f"-{parsed.name}.yaml"),
137 dict(parsed.spec.mutations),
138 )
141def submit_example(
142 parsed: ParsedExample,
143 manifest_path: Path,
144 *,
145 repo_root: Path,
146 region: str,
147 kubectl: KubectlRunner,
148) -> dict[str, Any]:
149 """Submit via the documented path; returns submission evidence."""
150 spec = parsed.spec
151 if spec.submission == SUBMIT_DIRECT:
152 args = ["gco", "jobs", "submit-direct", str(manifest_path), "-r", region]
153 elif spec.submission == SUBMIT_SQS:
154 args = ["gco", "jobs", "submit-sqs", str(manifest_path), "--region", region]
155 elif spec.submission == SUBMIT_API:
156 args = ["gco", "jobs", "submit", str(manifest_path), "--region", region]
157 elif spec.submission == DAG_RUN:
158 args = ["gco", "dag", "run", str(manifest_path), "-r", region]
159 elif spec.submission == KUBECTL_APPLY:
160 code, out, err = kubectl("apply", "-f", str(manifest_path))
161 if code != 0:
162 raise ExampleValidationError(f"kubectl apply failed: {err.strip()[:800]}")
163 return {"command": f"kubectl apply -f examples/{parsed.name}.yaml", "output": out.strip()}
164 else:
165 raise ExampleValidationError(f"No live submission for {spec.submission}")
167 timeout = 1800 if spec.submission == DAG_RUN else 600
168 # The CLI reads the session's kubeconfig, the file the tunnel is pinned in
169 # and kubectl uses; submit-direct shells out to kubectl with it.
170 environment = session_environment(kubectl)
172 def submit() -> tuple[int, str, str]:
173 return _run_cli(args, repo_root, timeout=timeout, env=environment)
175 if spec.submission == SUBMIT_DIRECT:
176 # submit-direct is the one submission that shells out to kubectl
177 # through this session's tunnel (a live run lost five examples to a
178 # stalled tunnel at exactly this step). It is repeated once the tunnel
179 # is back, but only while none of the example's Jobs exists: the CLI
180 # renames a second submission of a Job that is still running, which
181 # would start a duplicate the harness never cleans up.
182 code, out, err = through_tunnel(
183 kubectl, submit, safe_to_repeat=lambda: _no_job_landed(parsed, kubectl)
184 )
185 else:
186 code, out, err = submit()
187 if code != 0:
188 raise ExampleValidationError(
189 f"{' '.join(args[:3])} failed (exit {code}): {(err or out).strip()[:800]}"
190 )
191 return {"command": " ".join(args[:3]) + f" examples/{parsed.name}.yaml", "output": out[-1500:]}
194# --------------------------------------------------------------------------
195# success criteria
196# --------------------------------------------------------------------------
199def _workload_documents(parsed: ParsedExample, kinds: set[str]) -> list[dict[str, Any]]:
200 return [doc for doc in parsed.documents if doc.get("kind") in kinds]
203def _no_job_landed(parsed: ParsedExample, kubectl: KubectlRunner) -> bool:
204 """True when the API server answers NotFound for every Job the example defines."""
205 return all(
206 _job_status(
207 kubectl,
208 (doc.get("metadata") or {}).get("namespace", "gco-jobs"),
209 doc["metadata"]["name"],
210 )[0]
211 == "missing"
212 for doc in _workload_documents(parsed, {"Job"})
213 )
216def _job_status(kubectl: KubectlRunner, namespace: str, name: str) -> tuple[str, str]:
217 """``(state, detail)``: complete, failed, running, missing, or unreachable.
219 ``missing`` is only ever the API server answering NotFound. Any other
220 failed read — a stalled tunnel, a timeout — is ``unreachable`` with
221 kubectl's error, so a watcher keeps polling a Job it cannot see and a
222 cleanup check never takes a read it could not make for a Job that is gone
223 (a live run's watchers reported submitted Jobs as missing for 40 minutes
224 while every read failed with ``TLS handshake timeout``).
225 """
226 code, out, err = kubectl("get", "job", name, "-n", namespace, "-o", "json")
227 if code != 0:
228 if "(NotFound)" in err:
229 return "missing", ""
230 return "unreachable", (err or out).strip()[:300]
231 payload = json.loads(out)
232 for condition in payload.get("status", {}).get("conditions", []) or []:
233 if condition.get("type") == "Complete" and condition.get("status") == "True":
234 return "complete", ""
235 if condition.get("type") == "Failed" and condition.get("status") == "True":
236 return "failed", str(condition.get("message", ""))
237 return "running", ""
240def _pod_diagnostics(kubectl: KubectlRunner, namespace: str, selector: str) -> str:
241 _, out, _ = kubectl(
242 "get",
243 "pods",
244 "-n",
245 namespace,
246 "-l",
247 selector,
248 "-o",
249 "jsonpath={range .items[*]}{.metadata.name}={.status.phase} {end}",
250 )
251 return out.strip()
254def _job_admission_rejection(kubectl: KubectlRunner, namespace: str, name: str) -> str | None:
255 """Return the rejection message when the Job's pods are forbidden.
257 A LimitRange or ResourceQuota rejection never becomes a Job condition:
258 the controller retries pod creation forever, the Job stays podless, and
259 the only signal is ``FailedCreate ... forbidden`` namespace events.
260 Waiting the full example timeout on such a job is pure burn (observed
261 live: example-job validation run ex241-df723811, 40 minutes against the
262 old per-container GPU ceiling) — surface the event message immediately.
263 ResourceQuota rejections (``exceeded quota``) are the one retriable
264 shape: under parallel example submission the namespace quota is
265 transiently full, the Job controller retries pod creation, and the pods
266 land once peers finish. Never fail fast on those; each message is
267 evaluated separately so a transient quota event cannot mask a permanent
268 LimitRange rejection emitted for the same Job.
269 """
270 code, out, _ = kubectl(
271 "get",
272 "events",
273 "-n",
274 namespace,
275 "--field-selector",
276 f"involvedObject.kind=Job,involvedObject.name={name},reason=FailedCreate",
277 "-o",
278 'jsonpath={range .items[*]}{.message}{"\\n"}{end}',
279 timeout=60,
280 )
281 if code != 0:
282 return None
283 for message in out.splitlines():
284 if "forbidden" in message and "exceeded quota" not in message:
285 return message[-600:]
286 return None
289def wait_jobs_complete(
290 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
291) -> dict[str, Any]:
292 """Every batch/v1 Job in the example must reach Complete."""
293 jobs = [
294 ((doc.get("metadata") or {}).get("namespace", "gco-jobs"), doc["metadata"]["name"])
295 for doc in _workload_documents(parsed, {"Job"})
296 ]
297 if not jobs:
298 raise ExampleValidationError("spec says job-completes but the file defines no Jobs")
299 deadline = time.monotonic() + timeout
300 pending = dict.fromkeys(jobs, "unknown")
301 while time.monotonic() < deadline:
302 for namespace, name in jobs:
303 state, message = _job_status(kubectl, namespace, name)
304 pending[(namespace, name)] = f"{state} ({message})" if state == "unreachable" else state
305 if state == "failed":
306 _, logs, _ = kubectl(
307 "logs", f"job/{name}", "-n", namespace, "--tail", "40", timeout=60
308 )
309 raise ExampleValidationError(
310 f"Job {namespace}/{name} failed: {message} :: last logs: {logs[-800:]}"
311 )
312 rejection = _job_admission_rejection(kubectl, namespace, name)
313 if rejection is not None:
314 raise ExampleValidationError(
315 f"Job {namespace}/{name} pods are rejected at admission and can never "
316 f"run: {rejection}"
317 )
318 if all(state == "complete" for state in pending.values()):
319 return {"jobs": {f"{ns}/{name}": "complete" for (ns, name) in jobs}}
320 time.sleep(_POLL_SECONDS)
321 detail = ", ".join(f"{ns}/{name}={state}" for (ns, name), state in pending.items())
322 pods = _pod_diagnostics(kubectl, jobs[0][0], f"job-name={jobs[0][1]}")
323 raise ExampleValidationError(f"timeout after {timeout}s: {detail}; pods: {pods}")
326def wait_deployment_available(
327 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
328) -> dict[str, Any]:
329 """The Deployment must report Available and its Service must have endpoints."""
330 deployments = _workload_documents(parsed, {"Deployment"})
331 services = _workload_documents(parsed, {"Service"})
332 if not deployments:
333 raise ExampleValidationError("spec says deployment-available but no Deployment found")
334 namespace = deployments[0]["metadata"].get("namespace", "gco-inference")
335 name = deployments[0]["metadata"]["name"]
336 code, _, err = kubectl(
337 "wait",
338 f"deployment/{name}",
339 "-n",
340 namespace,
341 "--for",
342 "condition=Available",
343 f"--timeout={timeout}s",
344 timeout=timeout + 60,
345 )
346 if code != 0:
347 pods = _pod_diagnostics(kubectl, namespace, f"app={name}")
348 _, describe, _ = kubectl("describe", f"deployment/{name}", "-n", namespace, timeout=60)
349 raise ExampleValidationError(
350 f"Deployment {namespace}/{name} never became Available: {err.strip()[:300]}; "
351 f"pods: {pods}; describe tail: {describe[-600:]}"
352 )
353 evidence: dict[str, Any] = {"deployment": f"{namespace}/{name}=Available"}
354 if services:
355 service_name = services[0]["metadata"]["name"]
356 _, endpoints, _ = kubectl(
357 "get",
358 "endpoints",
359 service_name,
360 "-n",
361 namespace,
362 "-o",
363 "jsonpath={.subsets[*].addresses[*].ip}",
364 )
365 if not endpoints.strip():
366 raise ExampleValidationError(f"Service {namespace}/{service_name} has no endpoints")
367 evidence["service_endpoints"] = endpoints.strip()
368 return evidence
371def wait_raycluster_ready(
372 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
373) -> dict[str, Any]:
374 clusters = _workload_documents(parsed, {"RayCluster"})
375 namespace = clusters[0]["metadata"].get("namespace", "gco-jobs")
376 name = clusters[0]["metadata"]["name"]
377 min_workers = int(clusters[0]["spec"]["workerGroupSpecs"][0].get("minReplicas", 1))
378 deadline = time.monotonic() + timeout
379 while time.monotonic() < deadline:
380 code, out, _ = kubectl("get", "raycluster", name, "-n", namespace, "-o", "json")
381 if code == 0:
382 status = json.loads(out).get("status", {})
383 state = str(status.get("state", ""))
384 ready_workers = int(status.get("readyWorkerReplicas", 0) or 0)
385 if state.lower() == "ready" and ready_workers >= min_workers:
386 return {
387 "raycluster": f"{namespace}/{name}",
388 "state": state,
389 "ready_workers": ready_workers,
390 }
391 time.sleep(_POLL_SECONDS)
392 _, describe, _ = kubectl("describe", "raycluster", name, "-n", namespace, timeout=60)
393 raise ExampleValidationError(
394 f"RayCluster {namespace}/{name} not ready after {timeout}s; tail: {describe[-600:]}"
395 )
398def wait_vcjob_completes(
399 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
400) -> dict[str, Any]:
401 jobs = _workload_documents(parsed, {"Job"})
402 volcano_jobs = [
403 doc for doc in jobs if str(doc.get("apiVersion", "")).startswith("batch.volcano")
404 ]
405 namespace = volcano_jobs[0]["metadata"].get("namespace", "gco-jobs")
406 name = volcano_jobs[0]["metadata"]["name"]
407 deadline = time.monotonic() + timeout
408 while time.monotonic() < deadline:
409 code, out, _ = kubectl("get", "vcjob", name, "-n", namespace, "-o", "json")
410 if code == 0:
411 phase = str(json.loads(out).get("status", {}).get("state", {}).get("phase", ""))
412 if phase == "Completed":
413 return {"vcjob": f"{namespace}/{name}", "phase": phase}
414 if phase in {"Failed", "Aborted", "Terminated"}:
415 raise ExampleValidationError(f"vcjob {namespace}/{name} reached phase {phase}")
416 time.sleep(_POLL_SECONDS)
417 raise ExampleValidationError(f"vcjob {namespace}/{name} did not complete within {timeout}s")
420def wait_trainjob_completes(
421 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
422) -> dict[str, Any]:
423 """Kubeflow TrainJob must reach condition Complete (Failed is terminal).
425 Evidence carries the terminal condition plus the per-child-Job counts
426 from status.jobsStatus so the report shows the gang actually ran
427 (numNodes pods succeeded), not merely that a condition flipped.
428 """
429 trainjobs = _workload_documents(parsed, {"TrainJob"})
430 namespace = trainjobs[0]["metadata"].get("namespace", "gco-jobs")
431 name = trainjobs[0]["metadata"]["name"]
432 deadline = time.monotonic() + timeout
433 while time.monotonic() < deadline:
434 code, out, _ = kubectl("get", "trainjob", name, "-n", namespace, "-o", "json")
435 if code == 0:
436 status = json.loads(out).get("status", {}) or {}
437 jobs_status = status.get("jobsStatus", []) or []
438 for condition in status.get("conditions", []) or []:
439 if condition.get("status") != "True":
440 continue
441 if condition.get("type") == "Complete":
442 return {
443 "trainjob": f"{namespace}/{name}",
444 "condition": "Complete",
445 "jobsStatus": jobs_status,
446 }
447 if condition.get("type") == "Failed":
448 raise ExampleValidationError(
449 f"TrainJob {namespace}/{name} reached condition Failed: "
450 f"{condition.get('message', '')}"
451 )
452 time.sleep(_POLL_SECONDS)
453 _, describe, _ = kubectl("describe", "trainjob", name, "-n", namespace, timeout=60)
454 raise ExampleValidationError(
455 f"TrainJob {namespace}/{name} did not complete within {timeout}s; tail: {describe[-600:]}"
456 )
459def wait_scaledjob_scales(
460 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
461) -> dict[str, Any]:
462 """KEDA must spawn at least one Job for the ScaledJob from queue depth."""
463 scaled = _workload_documents(parsed, {"ScaledJob"})
464 namespace = scaled[0]["metadata"].get("namespace", "gco-jobs")
465 name = scaled[0]["metadata"]["name"]
466 deadline = time.monotonic() + timeout
467 while time.monotonic() < deadline:
468 code, out, _ = kubectl(
469 "get",
470 "jobs",
471 "-n",
472 namespace,
473 "-l",
474 f"scaledjob.keda.sh/name={name}",
475 "-o",
476 "jsonpath={.items[*].metadata.name}",
477 )
478 spawned = [item for item in out.split() if item]
479 if code == 0 and spawned:
480 return {"scaledjob": f"{namespace}/{name}", "spawned_jobs": spawned[:5]}
481 time.sleep(_POLL_SECONDS)
482 _, describe, _ = kubectl("describe", "scaledjob", name, "-n", namespace, timeout=60)
483 raise ExampleValidationError(
484 f"ScaledJob {namespace}/{name} spawned no Jobs within {timeout}s; tail: {describe[-600:]}"
485 )
488_SHA_RE = re.compile(r"[0-9a-f]{40}")
491def _conditions(payload: dict[str, Any]) -> list[dict[str, Any]]:
492 conditions = (payload.get("status") or {}).get("conditions") or []
493 return [condition for condition in conditions if isinstance(condition, dict)]
496def _condition_true(payload: dict[str, Any], condition_type: str) -> bool:
497 return any(
498 condition.get("type") == condition_type and condition.get("status") == "True"
499 for condition in _conditions(payload)
500 )
503def _condition_summary(payload: dict[str, Any]) -> str:
504 return "; ".join(
505 " ".join(
506 str(part)
507 for part in (
508 f"{condition.get('type')}={condition.get('status')}",
509 condition.get("reason", ""),
510 condition.get("message", ""),
511 )
512 if part
513 )
514 for condition in _conditions(payload)
515 )[:600]
518def _resource_ref(doc: dict[str, Any]) -> str:
519 """``kind.group`` for kubectl (``batchjob.kro.run``, ``queue.sqs.services.k8s.aws``)."""
520 group = str(doc.get("apiVersion", "")).rpartition("/")[0]
521 kind = str(doc.get("kind", "")).lower()
522 return f"{kind}.{group}" if group else kind
525def _get_json(kubectl: KubectlRunner, *args: str) -> dict[str, Any] | None:
526 code, out, _ = kubectl("get", *args, "-o", "json")
527 return json.loads(out) if code == 0 else None
530def wait_argocd_application_healthy(
531 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
532) -> dict[str, Any]:
533 """Every Application must be Synced + Healthy, and every Job it syncs Complete.
535 A 40-hex ``targetRevision`` (the harness pins the commit under validation)
536 must equal the revision Argo CD reports it synced. The Jobs are read from
537 the Application's Git path in this checkout and must appear among the
538 resources Argo CD manages, so the evidence proves Argo CD applied them.
539 """
540 applications = _workload_documents(parsed, {"Application"})
541 if not applications:
542 raise ExampleValidationError(
543 "spec says argocd-app-healthy but the file defines no Application"
544 )
545 repo_root = parsed.path.parent.parent
546 deadline = time.monotonic() + timeout
547 synced: dict[str, dict[str, Any]] = {}
548 managed: set[tuple[str, str, str]] = set()
549 for doc in applications:
550 namespace = doc["metadata"].get("namespace", "argocd")
551 name = doc["metadata"]["name"]
552 last = "Application not found"
553 while True:
554 app = _get_json(kubectl, "application", name, "-n", namespace)
555 if app is not None:
556 status = app.get("status") or {}
557 sync = status.get("sync") or {}
558 health = status.get("health") or {}
559 operation = status.get("operationState") or {}
560 if operation.get("phase") in {"Failed", "Error"}:
561 raise ExampleValidationError(
562 f"Application {namespace}/{name} sync {operation['phase']}: "
563 f"{str(operation.get('message', ''))[:400]}; {_condition_summary(app)}"
564 )
565 if health.get("status") == "Degraded":
566 raise ExampleValidationError(
567 f"Application {namespace}/{name} is Degraded: "
568 f"{str(health.get('message', ''))[:300]}; {_condition_summary(app)}"
569 )
570 if sync.get("status") == "Synced" and health.get("status") == "Healthy":
571 target = str(
572 ((app.get("spec") or {}).get("source") or {}).get("targetRevision")
573 )
574 revision = str(sync.get("revision", ""))
575 if _SHA_RE.fullmatch(target) and revision != target:
576 raise ExampleValidationError(
577 f"Application {namespace}/{name} synced revision {revision!r}, "
578 f"not the pinned {target!r}"
579 )
580 synced[f"{namespace}/{name}"] = {"revision": revision, "resources": 0}
581 for resource in status.get("resources") or []:
582 managed.add(
583 (
584 str(resource.get("kind", "")),
585 str(resource.get("namespace", "")),
586 str(resource.get("name", "")),
587 )
588 )
589 synced[f"{namespace}/{name}"]["resources"] += 1
590 break
591 last = (
592 f"sync={sync.get('status', '')} health={health.get('status', '')} "
593 f"operation={operation.get('phase', '')}; {_condition_summary(app)}"
594 )
595 if time.monotonic() >= deadline:
596 raise ExampleValidationError(
597 f"Application {namespace}/{name} not Synced/Healthy within {timeout}s: {last}"
598 )
599 time.sleep(_POLL_SECONDS)
600 jobs: dict[str, str] = {}
601 for doc in applications:
602 for item in application_source_documents(repo_root, doc):
603 if item.get("kind") != "Job":
604 continue
605 namespace = item["metadata"]["namespace"]
606 name = item["metadata"]["name"]
607 if ("Job", namespace, name) not in managed:
608 raise ExampleValidationError(
609 f"Job {namespace}/{name} from the Git path is not among the resources "
610 f"Argo CD manages for {doc['metadata']['name']}"
611 )
612 state, message = _job_status(kubectl, namespace, name)
613 if state != "complete":
614 raise ExampleValidationError(
615 f"Application is Healthy but Job {namespace}/{name} is {state} {message}".strip()
616 )
617 jobs[f"{namespace}/{name}"] = "complete"
618 if not jobs:
619 raise ExampleValidationError("the Application syncs no Job from this repository")
620 return {"applications": synced, "jobs": jobs}
623def _instance_summary(kubectl: KubectlRunner, doc: dict[str, Any], namespace: str) -> str:
624 instance = _get_json(kubectl, _resource_ref(doc), doc["metadata"]["name"], "-n", namespace)
625 if instance is None:
626 return "instance not found"
627 state = (instance.get("status") or {}).get("state")
628 conditions = _condition_summary(instance)
629 return " ".join(part for part in (f"state={state}" if state else "", conditions) if part) or (
630 "no status yet"
631 )
634def wait_composed_jobs_complete(
635 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
636) -> dict[str, Any]:
637 """Each instance's composed Job — named after the instance — must reach Complete.
639 The example file holds only the instance (a kro or Crossplane custom
640 resource), so the Job's existence already proves the composition ran; the
641 instance's own state and conditions are reported with it (and on timeout).
642 """
643 instances = [doc for doc in parsed.documents if (doc.get("metadata") or {}).get("name")]
644 if not instances:
645 raise ExampleValidationError(
646 "spec says composed-job-completes but the file defines nothing"
647 )
648 deadline = time.monotonic() + timeout
649 results: dict[str, dict[str, str]] = {}
650 for doc in instances:
651 namespace = doc["metadata"].get("namespace", "gco-jobs")
652 name = doc["metadata"]["name"]
653 while True:
654 state, message = _job_status(kubectl, namespace, name)
655 if state == "complete":
656 results[f"{namespace}/{name}"] = {
657 "job": "complete",
658 "instance": _instance_summary(kubectl, doc, namespace),
659 }
660 break
661 if state == "failed":
662 _, logs, _ = kubectl(
663 "logs", f"job/{name}", "-n", namespace, "--tail", "40", timeout=60
664 )
665 raise ExampleValidationError(
666 f"composed Job {namespace}/{name} failed: {message} :: last logs: {logs[-800:]}"
667 )
668 rejection = _job_admission_rejection(kubectl, namespace, name)
669 if rejection is not None:
670 raise ExampleValidationError(
671 f"composed Job {namespace}/{name} pods are rejected at admission: {rejection}"
672 )
673 if time.monotonic() >= deadline:
674 raise ExampleValidationError(
675 f"composed Job {namespace}/{name} is {state} after {timeout}s; "
676 f"{_resource_ref(doc)}: {_instance_summary(kubectl, doc, namespace)}"
677 )
678 time.sleep(_POLL_SECONDS)
679 return {"composed_jobs": results}
682def wait_ack_resources_synced(
683 parsed: ParsedExample, kubectl: KubectlRunner, *, timeout: int
684) -> dict[str, Any]:
685 """Every ACK resource must report ``ACK.ResourceSynced=True`` (``ACK.Terminal`` fails fast).
687 ACK sets ResourceSynced only after reading the resource back from AWS, so
688 the evidence (ARN, URL fields) is AWS's own answer. ``ACK.Recoverable``
689 (an access denial, say) keeps the wait going and is reported on timeout.
690 """
691 resources = [
692 doc for doc in parsed.documents if ".services.k8s.aws/" in str(doc.get("apiVersion", ""))
693 ]
694 if not resources:
695 raise ExampleValidationError("spec says ack-resource-synced but the file has no ACK kind")
696 deadline = time.monotonic() + timeout
697 results: dict[str, dict[str, Any]] = {}
698 for doc in resources:
699 namespace = doc["metadata"].get("namespace", "gco-jobs")
700 name = doc["metadata"]["name"]
701 ref = _resource_ref(doc)
702 last = "not found"
703 while True:
704 payload = _get_json(kubectl, ref, name, "-n", namespace)
705 if payload is not None:
706 if _condition_true(payload, "ACK.Terminal"):
707 raise ExampleValidationError(
708 f"{ref} {namespace}/{name} is terminal: {_condition_summary(payload)}"
709 )
710 if _condition_true(payload, "ACK.ResourceSynced"):
711 status = payload.get("status") or {}
712 results[f"{namespace}/{name}"] = {
713 "arn": str((status.get("ackResourceMetadata") or {}).get("arn", "")),
714 **{
715 key: value
716 for key, value in status.items()
717 if key.lower().endswith("url") and isinstance(value, str)
718 },
719 }
720 break
721 last = _condition_summary(payload) or "no conditions yet"
722 if time.monotonic() >= deadline:
723 raise ExampleValidationError(
724 f"{ref} {namespace}/{name} not synced within {timeout}s: {last}"
725 )
726 time.sleep(_POLL_SECONDS)
727 return {"ack_resources": results}
730CRITERIA_WAITERS = {
731 JOB_COMPLETES: wait_jobs_complete,
732 DEPLOYMENT_AVAILABLE: wait_deployment_available,
733 RAYCLUSTER_READY: wait_raycluster_ready,
734 VCJOB_COMPLETES: wait_vcjob_completes,
735 SCALEDJOB_SCALES: wait_scaledjob_scales,
736 TRAINJOB_COMPLETES: wait_trainjob_completes,
737 ARGOCD_APP_HEALTHY: wait_argocd_application_healthy,
738 COMPOSED_JOB_COMPLETES: wait_composed_jobs_complete,
739 ACK_RESOURCE_SYNCED: wait_ack_resources_synced,
740}
743# --------------------------------------------------------------------------
744# cleanup
745# --------------------------------------------------------------------------
748def cleanup_example(
749 parsed: ParsedExample, manifest_path: Path, kubectl: KubectlRunner
750) -> dict[str, Any]:
751 """Delete everything the example created and verify it is gone."""
752 if parsed.spec.submission == DAG_RUN:
753 # A DAG example's file is a pipeline SPEC, not a Kubernetes manifest
754 # (kubectl cannot decode it — observed live in run ex241-4bf01801);
755 # what actually ran on the cluster are the step manifests it names.
756 repo_root = parsed.path.parent.parent
757 deleted: list[str] = []
758 for document in parsed.documents:
759 for step in document.get("steps", []):
760 step_manifest = repo_root / str(step.get("manifest", ""))
761 code, out, err = kubectl(
762 "delete",
763 "-f",
764 str(step_manifest),
765 "--ignore-not-found",
766 "--wait=true",
767 timeout=300,
768 )
769 if code != 0:
770 raise ExampleValidationError(
771 f"cleanup failed for {parsed.name} step "
772 f"{step.get('name', '?')}: {err.strip()[:500]}"
773 )
774 deleted.extend(line for line in out.strip().splitlines() if line)
775 return {"deleted": deleted[:20]}
776 code, out, err = kubectl(
777 "delete", "-f", str(manifest_path), "--ignore-not-found", "--wait=true", timeout=300
778 )
779 if code != 0:
780 raise ExampleValidationError(f"cleanup failed for {parsed.name}: {err.strip()[:500]}")
781 result: dict[str, Any] = {"deleted": [line for line in out.strip().splitlines() if line][:20]}
782 derived = _derived_jobs(parsed)
783 if derived:
784 result["derived_jobs_removed"] = _wait_jobs_gone(kubectl, derived, timeout=180)
785 return result
788def _derived_jobs(parsed: ParsedExample) -> list[tuple[str, str]]:
789 """Jobs the example creates indirectly: synced by Argo CD or composed by kro/Crossplane.
791 Deleting the example's own objects deletes these asynchronously (the
792 Application's resources finalizer, the instance's composition), so cleanup
793 waits for them too: a lingering Job would hold namespace quota.
794 """
795 if parsed.spec.criteria == ARGOCD_APP_HEALTHY:
796 repo_root = parsed.path.parent.parent
797 return [
798 (item["metadata"]["namespace"], item["metadata"]["name"])
799 for doc in _workload_documents(parsed, {"Application"})
800 for item in application_source_documents(repo_root, doc)
801 if item.get("kind") == "Job"
802 ]
803 if parsed.spec.criteria == COMPOSED_JOB_COMPLETES:
804 return [
805 ((doc.get("metadata") or {}).get("namespace", "gco-jobs"), doc["metadata"]["name"])
806 for doc in parsed.documents
807 if (doc.get("metadata") or {}).get("name")
808 ]
809 return []
812def _wait_jobs_gone(
813 kubectl: KubectlRunner, jobs: list[tuple[str, str]], *, timeout: int
814) -> list[str]:
815 deadline = time.monotonic() + timeout
816 remaining = list(jobs)
817 while True:
818 states = {job: _job_status(kubectl, *job)[0] for job in remaining}
819 remaining = [job for job in remaining if states[job] != "missing"]
820 if not remaining:
821 return [f"{namespace}/{name}" for namespace, name in jobs]
822 if time.monotonic() >= deadline:
823 # "unreachable" is named as such: a Job the reads could not see
824 # is not proven gone, and it is not proven present either.
825 raise ExampleValidationError(
826 "derived Job(s) still present after cleanup: "
827 + ", ".join(
828 f"{namespace}/{name} ({states[(namespace, name)]})"
829 for namespace, name in remaining
830 )
831 )
832 time.sleep(_POLL_SECONDS)
835# --------------------------------------------------------------------------
836# setup drivers (spec.setup_driver)
837# --------------------------------------------------------------------------
840@dataclass
841class KedaDemoQueue:
842 """Disposable SQS queue backing the KEDA scaling demonstration.
844 Implements the example's documented prerequisites: a demo queue with
845 synthetic messages and read-only queue-metric access for the KEDA
846 operator (granted with a queue policy, never by touching IAM roles).
847 """
849 session: Any
850 region: str
851 run_id: str
852 queue_url: str = ""
853 queue_arn: str = ""
855 def _sqs(self) -> Any:
856 # boto3 Session.client() is not thread-safe; examples may run in
857 # parallel threads. The returned client is safe to use concurrently.
858 with BOTO_CLIENT_LOCK:
859 return self.session.client("sqs", region_name=self.region)
861 def create(self, operator_role_arn: str) -> dict[str, Any]:
862 sqs = self._sqs()
863 name = f"gco-keda-demo-{self.run_id}"[:80]
864 self.queue_url = sqs.create_queue(QueueName=name)["QueueUrl"]
865 attrs = sqs.get_queue_attributes(QueueUrl=self.queue_url, AttributeNames=["QueueArn"])
866 self.queue_arn = attrs["Attributes"]["QueueArn"]
867 policy = {
868 "Version": "2012-10-17",
869 "Statement": [
870 {
871 "Sid": "KedaOperatorQueueMetrics",
872 "Effect": "Allow",
873 "Principal": {"AWS": operator_role_arn},
874 "Action": ["sqs:GetQueueAttributes", "sqs:GetQueueUrl"],
875 "Resource": self.queue_arn,
876 }
877 ],
878 }
879 sqs.set_queue_attributes(QueueUrl=self.queue_url, Attributes={"Policy": json.dumps(policy)})
880 for index in range(10):
881 sqs.send_message(QueueUrl=self.queue_url, MessageBody=f"demo-{index}")
882 return {"queue_arn": self.queue_arn, "seeded_messages": 10}
884 def destroy(self) -> None:
885 if self.queue_url:
886 self._sqs().delete_queue(QueueUrl=self.queue_url)
889@dataclass
890class VectorDemoCorpus:
891 """Demo-corpus precondition for the vector-search example, fully reverted.
893 ``create`` runs the example's documented prerequisite verbatim —
894 ``gco vector ingest --demo --wait`` — and records exactly which corpus
895 objects the CLI uploaded. ``destroy`` reverts precisely those: the S3
896 objects (whose upload is what triggered ingestion) and every DynamoDB
897 chunk item whose ``source`` is one of the recorded keys, resolved with
898 the same filtered-Scan shape the CLI's ingest wait uses (the table has
899 no by-source key schema; corpora are document-scale). Scoping deletion
900 to the recorded keys means a pre-existing user corpus in the same table
901 is never touched.
902 """
904 repo_root: Path
905 session: Any
906 region: str
907 uploaded: list[str] = field(default_factory=list)
908 bucket: str = ""
910 def create(self) -> dict[str, Any]:
911 code, out, err = _run_cli(
912 ["gco", "vector", "ingest", "--demo", "--wait", "--output", "json"],
913 self.repo_root,
914 timeout=900,
915 )
916 if code != 0:
917 raise ExampleValidationError(
918 f"gco vector ingest --demo --wait failed (exit {code}): "
919 f"{(err or out).strip()[:800]}"
920 )
921 try:
922 summary = json.loads(out)
923 except ValueError as exc:
924 raise ExampleValidationError(
925 f"gco vector ingest emitted non-JSON output: {out[:400]}"
926 ) from exc
927 self.bucket = str(summary.get("bucket", ""))
928 self.uploaded = [str(key) for key in summary.get("uploaded", [])]
929 if not self.bucket or not self.uploaded:
930 raise ExampleValidationError(
931 f"ingest summary carried no bucket/keys to revert later: {out[:400]}"
932 )
933 return {
934 "command": "gco vector ingest --demo --wait",
935 "bucket": self.bucket,
936 "uploaded": self.uploaded,
937 "chunks_by_source": summary.get("chunks_by_source", {}),
938 }
940 def destroy(self) -> None:
941 if not self.uploaded:
942 return
943 from cli.vector_store import VectorStoreClient
945 client = VectorStoreClient(query_region=self.region)
946 table_name = client._resolve_table_name()
947 bucket_name, bucket_region = client._resolve_bucket()
948 if bucket_name != self.bucket:
949 # Fail loudly rather than delete from a bucket other than the
950 # one create() actually uploaded to.
951 raise ExampleValidationError(
952 f"corpus bucket changed between ingest ({self.bucket}) and "
953 f"revert ({bucket_name}); refusing to delete"
954 )
955 with BOTO_CLIENT_LOCK:
956 dynamodb = self.session.client("dynamodb", region_name=self.region)
957 s3 = self.session.client("s3", region_name=bucket_region)
959 # Chunk items first (their source keys reference the S3 objects),
960 # then the objects themselves. Deleting the objects does NOT
961 # un-ingest — the notification only fires on creates — hence the
962 # explicit item sweep.
963 for key in self.uploaded:
964 doc_ids: list[str] = []
965 scan_kwargs: dict[str, Any] = {
966 "TableName": table_name,
967 "FilterExpression": "#source = :source",
968 "ExpressionAttributeNames": {"#source": "source"},
969 "ExpressionAttributeValues": {":source": {"S": key}},
970 "ProjectionExpression": "doc_id",
971 }
972 while True:
973 page = dynamodb.scan(**scan_kwargs)
974 doc_ids.extend(
975 item["doc_id"]["S"] for item in page.get("Items", []) if "doc_id" in item
976 )
977 last_key = page.get("LastEvaluatedKey")
978 if not last_key:
979 break
980 scan_kwargs["ExclusiveStartKey"] = last_key
981 for start in range(0, len(doc_ids), 25):
982 batch = doc_ids[start : start + 25]
983 dynamodb.batch_write_item(
984 RequestItems={
985 table_name: [
986 {"DeleteRequest": {"Key": {"doc_id": {"S": doc_id}}}}
987 for doc_id in batch
988 ]
989 }
990 )
991 s3.delete_object(Bucket=bucket_name, Key=key)
994def wait_trainer_runtime_ready(kubectl: KubectlRunner, *, timeout: int = 300) -> dict[str, Any]:
995 """Wait until the TrainJob CRD is served and the shipped runtime exists.
997 The trainer chart installs the CRDs and controller; the post-Helm
998 kubectl pass applies the torch-distributed ClusterTrainingRuntime the
999 example's ``runtimeRef`` names. Both are deploy-time artifacts, so this
1000 is a readiness wait, not created state — nothing to revert.
1001 """
1002 deadline = time.monotonic() + timeout
1003 last_error = ""
1004 while True:
1005 code, _, err = kubectl("get", "crd", "trainjobs.trainer.kubeflow.org")
1006 if code != 0:
1007 last_error = f"TrainJob CRD not present: {err.strip()[:300]}"
1008 else:
1009 code, out, err = kubectl(
1010 "get", "clustertrainingruntime", "torch-distributed", "-o", "json"
1011 )
1012 if code == 0:
1013 runtime = json.loads(out)
1014 return {
1015 "crd": "trainjobs.trainer.kubeflow.org",
1016 "runtime": runtime["metadata"]["name"],
1017 "runtime_created": runtime["metadata"].get("creationTimestamp", ""),
1018 }
1019 last_error = f"torch-distributed runtime not present: {err.strip()[:300]}"
1020 if time.monotonic() >= deadline:
1021 break
1022 time.sleep(_POLL_SECONDS)
1023 raise ExampleValidationError(
1024 f"Kubeflow Trainer runtime not ready within {timeout}s — is "
1025 f"helm.kubeflow_trainer enabled? Last error: {last_error}"
1026 )
1029#: The CA bundle trust-manager publishes for MLflow's clients
1030#: (post-helm-mlflow-tls.yaml); the example mounts it to verify the server.
1031MLFLOW_CA_BUNDLE_NAMESPACE = "gco-jobs"
1032MLFLOW_CA_BUNDLE_CONFIGMAP = "gco-internal-ca"
1035def _mlflow_ca_bundle_state(kubectl: KubectlRunner) -> str:
1036 """Return why the MLflow clients' CA bundle is not usable yet, or ``""``."""
1037 reference = f"{MLFLOW_CA_BUNDLE_NAMESPACE}/{MLFLOW_CA_BUNDLE_CONFIGMAP}"
1038 code, out, err = kubectl(
1039 "get",
1040 "configmap",
1041 MLFLOW_CA_BUNDLE_CONFIGMAP,
1042 "-n",
1043 MLFLOW_CA_BUNDLE_NAMESPACE,
1044 "-o",
1045 "json",
1046 )
1047 if code != 0:
1048 return f"CA bundle ConfigMap {reference} not found: {err.strip()[:300]}"
1049 data = json.loads(out).get("data") or {}
1050 if not str(data.get("ca.crt", "")).strip():
1051 return f"CA bundle ConfigMap {reference} has no ca.crt yet"
1052 return ""
1055def wait_mlflow_ready(kubectl: KubectlRunner, *, timeout: int = 600) -> dict[str, Any]:
1056 """Wait until the MLflow tracking server is Available and its clients' CA is published.
1058 The example's client job fails its read-back (or hangs on connect) if
1059 it races the server's first rollout. It also mounts the
1060 ``gco-jobs/gco-internal-ca`` ConfigMap, the CA certificate it verifies
1061 the server's HTTPS listener with, which trust-manager writes from the
1062 ``gco-internal-ca`` Bundle once the post-Helm pass has created both; a
1063 pod created before that waits in ContainerCreating. Readiness wait only;
1064 nothing to revert.
1065 """
1066 deadline = time.monotonic() + timeout
1067 last_state = ""
1068 while True:
1069 code, out, err = kubectl("get", "deployment", "mlflow", "-n", "monitoring", "-o", "json")
1070 if code != 0:
1071 last_state = f"mlflow Deployment not found: {err.strip()[:300]}"
1072 else:
1073 payload = json.loads(out)
1074 conditions = payload.get("status", {}).get("conditions", []) or []
1075 available = any(
1076 condition.get("type") == "Available" and condition.get("status") == "True"
1077 for condition in conditions
1078 )
1079 if not available:
1080 last_state = f"conditions: {json.dumps(conditions)[:400]}"
1081 else:
1082 last_state = _mlflow_ca_bundle_state(kubectl)
1083 if not last_state:
1084 return {
1085 "deployment": "monitoring/mlflow",
1086 "ready_replicas": payload.get("status", {}).get("readyReplicas", 0),
1087 "ca_bundle": f"{MLFLOW_CA_BUNDLE_NAMESPACE}/{MLFLOW_CA_BUNDLE_CONFIGMAP}",
1088 }
1089 if time.monotonic() >= deadline:
1090 break
1091 time.sleep(_POLL_SECONDS)
1092 raise ExampleValidationError(
1093 f"MLflow tracking server not ready within {timeout}s — is "
1094 f"cluster_observability.mlflow enabled? Last state: {last_state}"
1095 )
1098def _wait_until(probe: Any, *, deadline: float, what: str, hint: str = "") -> Any:
1099 """Poll ``probe()`` (returns a truthy result or a falsy one plus state) until the deadline."""
1100 last = ""
1101 while True:
1102 result, state = probe()
1103 if result:
1104 return result
1105 last = state or last
1106 if time.monotonic() >= deadline:
1107 suffix = f" ({hint})" if hint else ""
1108 raise ExampleValidationError(
1109 f"timed out waiting for {what}. Last state: {last}{suffix}"
1110 )
1111 time.sleep(_POLL_SECONDS)
1114def _crd_probe(kubectl: KubectlRunner, crd: str) -> Any:
1115 def probe() -> tuple[bool, str]:
1116 code, _, err = kubectl("get", "crd", crd)
1117 return code == 0, f"CRD {crd} not present: {err.strip()[:200]}"
1119 return probe
1122def _condition_probe(kubectl: KubectlRunner, resource: str, name: str, condition: str) -> Any:
1123 def probe() -> tuple[bool, str]:
1124 payload = _get_json(kubectl, resource, name)
1125 if payload is None:
1126 return False, f"{resource}/{name} not found"
1127 return _condition_true(payload, condition), _condition_summary(payload) or "no conditions"
1129 return probe
1132def _served_probe(kubectl: KubectlRunner, resource: str) -> Any:
1133 def probe() -> tuple[bool, str]:
1134 code, _, err = kubectl("get", resource, "-A", "-o", "name")
1135 return code == 0, f"{resource} not served yet: {err.strip()[:200]}"
1137 return probe
1140def wait_argocd_ready(kubectl: KubectlRunner, *, timeout: int = 600) -> dict[str, Any]:
1141 """Wait until Argo CD can sync: the fenced project exists, controller and repo server run.
1143 All three are deploy-time artifacts (the argo-cd chart and the post-Helm
1144 ``post-helm-argocd-access.yaml``), so this is a readiness wait with
1145 nothing to revert.
1146 """
1147 deadline = time.monotonic() + timeout
1149 def probe() -> tuple[dict[str, Any] | None, str]:
1150 if _get_json(kubectl, "appproject", "gco-tenants", "-n", "argocd") is None:
1151 return None, "AppProject argocd/gco-tenants not found"
1152 controller = _get_json(
1153 kubectl, "statefulset", "argocd-application-controller", "-n", "argocd"
1154 )
1155 ready = int(((controller or {}).get("status") or {}).get("readyReplicas") or 0)
1156 if ready < 1:
1157 return None, "argocd-application-controller has no ready replica"
1158 repo_server = _get_json(kubectl, "deployment", "argocd-repo-server", "-n", "argocd")
1159 if repo_server is None or not _condition_true(repo_server, "Available"):
1160 return None, "argocd-repo-server is not Available"
1161 return {
1162 "project": "argocd/gco-tenants",
1163 "application_controller_ready": ready,
1164 "repo_server": "Available",
1165 }, ""
1167 ready: dict[str, Any] = _wait_until(
1168 probe, deadline=deadline, what="Argo CD", hint="is helm.argocd enabled for this deploy?"
1169 )
1170 return ready
1173@dataclass
1174class CompanionApi:
1175 """A companion API definition applied before an instance example and deleted after it.
1177 ``flavor`` ``kro`` applies a ResourceGraphDefinition (after the kro
1178 capability's CRD exists) and waits for it to be Active; ``crossplane``
1179 waits for every composition function the Compositions reference to be
1180 Healthy, applies the XRD + Composition and waits for the XRD to be
1181 Established. Either way the new API must be served before the instance
1182 is applied, so ``kubectl apply`` never races discovery.
1183 """
1185 flavor: str
1186 path: Path
1187 kubectl: KubectlRunner
1188 applied: bool = False
1190 def _documents(self) -> list[dict[str, Any]]:
1191 return [
1192 doc
1193 for doc in yaml.safe_load_all(self.path.read_text(encoding="utf-8"))
1194 if isinstance(doc, dict)
1195 ]
1197 def create(self, *, timeout: int = 600) -> dict[str, Any]:
1198 if self.flavor not in {"kro", "crossplane"}:
1199 raise ExampleValidationError(f"unknown companion flavor {self.flavor!r}")
1200 deadline = time.monotonic() + timeout
1201 documents = self._documents()
1202 prerequisites: list[str] = []
1203 if self.flavor == "kro":
1204 crd = "resourcegraphdefinitions.kro.run"
1205 _wait_until(
1206 _crd_probe(self.kubectl, crd),
1207 deadline=deadline,
1208 what="the kro capability",
1209 hint="is eks_capabilities.kro enabled for this deploy?",
1210 )
1211 prerequisites.append(crd)
1212 else:
1213 functions = sorted(
1214 {
1215 str((step.get("functionRef") or {}).get("name", ""))
1216 for doc in documents
1217 if doc.get("kind") == "Composition"
1218 for step in (doc.get("spec") or {}).get("pipeline") or []
1219 }
1220 )
1221 for function in functions:
1222 _wait_until(
1223 _condition_probe(
1224 self.kubectl, "function.pkg.crossplane.io", function, "Healthy"
1225 ),
1226 deadline=deadline,
1227 what=f"Crossplane function {function}",
1228 hint="is helm.crossplane enabled for this deploy?",
1229 )
1230 prerequisites.append(f"function/{function}=Healthy")
1231 code, out, err = self.kubectl("apply", "-f", str(self.path))
1232 if code != 0:
1233 raise ExampleValidationError(
1234 f"kubectl apply -f examples/{self.path.name} failed: {err.strip()[:600]}"
1235 )
1236 self.applied = True
1237 served: list[str] = []
1238 for doc in documents:
1239 name = str((doc.get("metadata") or {}).get("name", ""))
1240 spec = doc.get("spec") or {}
1241 if self.flavor == "kro" and doc.get("kind") == "ResourceGraphDefinition":
1243 def rgd_probe(name: str = name) -> tuple[bool, str]:
1244 payload = _get_json(self.kubectl, "resourcegraphdefinition", name)
1245 state = str(((payload or {}).get("status") or {}).get("state", ""))
1246 summary = _condition_summary(payload or {})
1247 return state == "Active", f"state={state or 'unknown'} {summary}".strip()
1249 _wait_until(rgd_probe, deadline=deadline, what=f"ResourceGraphDefinition {name}")
1250 schema = spec.get("schema") or {}
1251 resource = f"{str(schema.get('kind', '')).lower()}.{schema.get('group', 'kro.run')}"
1252 elif self.flavor == "crossplane" and doc.get("kind") == "CompositeResourceDefinition":
1253 _wait_until(
1254 _condition_probe(
1255 self.kubectl,
1256 "compositeresourcedefinition.apiextensions.crossplane.io",
1257 name,
1258 "Established",
1259 ),
1260 deadline=deadline,
1261 what=f"CompositeResourceDefinition {name}",
1262 )
1263 names = spec.get("names") or {}
1264 resource = f"{str(names.get('kind', '')).lower()}.{spec.get('group', '')}"
1265 else:
1266 continue
1267 _wait_until(
1268 _served_probe(self.kubectl, resource), deadline=deadline, what=f"the {resource} API"
1269 )
1270 served.append(resource)
1271 return {
1272 "prerequisites": prerequisites,
1273 "applied": [line for line in out.strip().splitlines() if line][:20],
1274 "served": served,
1275 }
1277 def destroy(self) -> dict[str, Any]:
1278 if not self.applied:
1279 return {"deleted": []}
1280 code, out, err = self.kubectl(
1281 "delete", "-f", str(self.path), "--ignore-not-found", "--wait=true", timeout=300
1282 )
1283 if code != 0:
1284 raise ExampleValidationError(
1285 f"deleting companion examples/{self.path.name} failed: {err.strip()[:500]}"
1286 )
1287 self.applied = False
1288 return {"deleted": [line for line in out.strip().splitlines() if line][:20]}
1291@dataclass
1292class AckSqsQueues:
1293 """AWS-side proof for the ACK SQS example.
1295 ``wait_ready`` waits for the ACK capability's Queue CRD; after the
1296 example syncs, ``verify_created`` resolves each queue in SQS directly;
1297 after cleanup, ``verify_deleted`` requires SQS to stop resolving it — ACK
1298 deletes the AWS queue when the object goes, so a surviving queue is a leak.
1299 """
1301 session: Any
1302 region: str
1304 def _sqs(self) -> Any:
1305 with BOTO_CLIENT_LOCK:
1306 return self.session.client("sqs", region_name=self.region)
1308 @staticmethod
1309 def queue_names(parsed: ParsedExample) -> list[str]:
1310 return [
1311 str((doc.get("spec") or {}).get("queueName", ""))
1312 for doc in parsed.documents
1313 if doc.get("kind") == "Queue"
1314 ]
1316 def wait_ready(self, kubectl: KubectlRunner, *, timeout: int = 600) -> dict[str, Any]:
1317 crd = "queues.sqs.services.k8s.aws"
1318 _wait_until(
1319 _crd_probe(kubectl, crd),
1320 deadline=time.monotonic() + timeout,
1321 what="the ACK SQS controller",
1322 hint="is eks_capabilities.ack enabled for this deploy?",
1323 )
1324 return {"crd": crd}
1326 def verify_created(self, parsed: ParsedExample) -> dict[str, Any]:
1327 sqs = self._sqs()
1328 urls: dict[str, str] = {}
1329 for name in self.queue_names(parsed):
1330 try:
1331 urls[name] = str(sqs.get_queue_url(QueueName=name)["QueueUrl"])
1332 except sqs.exceptions.QueueDoesNotExist as exc:
1333 raise ExampleValidationError(
1334 f"ACK reports queue {name} synced but SQS does not resolve it"
1335 ) from exc
1336 return {"sqs_queue_urls": urls}
1338 def verify_deleted(self, parsed: ParsedExample, *, timeout: int = 180) -> dict[str, Any]:
1339 sqs = self._sqs()
1340 deadline = time.monotonic() + timeout
1341 for name in self.queue_names(parsed):
1343 def gone(name: str = name) -> tuple[bool, str]:
1344 try:
1345 sqs.get_queue_url(QueueName=name)
1346 except sqs.exceptions.QueueDoesNotExist:
1347 return True, ""
1348 return False, f"SQS still resolves {name}"
1350 _wait_until(gone, deadline=deadline, what=f"deletion of SQS queue {name}")
1351 return {"sqs_queues_deleted": self.queue_names(parsed)}
1354#: Setup drivers _run_one_example knows how to dispatch, by spec.setup_driver
1355#: name. Kept as an explicit registry so a spec naming a driver that does not
1356#: exist fails the registry pin test, not a live run.
1357KNOWN_SETUP_DRIVERS = frozenset(
1358 {
1359 "keda-demo-queue",
1360 "vector-demo-corpus",
1361 "trainer-runtime-ready",
1362 "mlflow-ready",
1363 "argocd-revision-pin",
1364 "kro-api",
1365 "crossplane-api",
1366 "ack-sqs",
1367 }
1368)