Coverage for scripts / live_release_validation / runner.py: 100.00%
298 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Dependency-aware runner with checkpointing and guaranteed cleanup."""
3from __future__ import annotations
5import json
6import os
7import signal
8import time
9import traceback
10from pathlib import Path
11from typing import Any, Literal
13from cli.aws_client import GCOAWSClient
14from cli.config import GCOConfig
15from cli.jobs import JobManager
16from cli.stacks import StackManager
18from .actions import action_final_inventory, destroy_deployment
19from .aws_session import ThrottleResilientSession
20from .models import (
21 ActionResult,
22 RunCheckpoint,
23 RunContext,
24 RunSettings,
25 ValidationReport,
26 atomic_write_json,
27 ensure_private_run_directory,
28 utc_now,
29)
30from .registry import ActionDefinition, build_action_registry
33class _LiveValidationSignal(BaseException):
34 """Controlled interruption raised by SIGTERM/SIGHUP handlers."""
36 def __init__(self, signum: int):
37 self.signum = signum
38 self.signal_name = signal.Signals(signum).name
39 super().__init__(f"Received {self.signal_name}")
42def require_local_execution() -> None:
43 """Reject GitHub Actions before creating checkpoints or AWS clients.
45 One verified exception: a run explicitly pointed at a local AWS emulator
46 (see ``emulator.py``) may execute in CI. The emulator proof runs first
47 and fails closed, so CI still cannot reach a real AWS account through
48 this path — a real endpoint fails the URL rules, and real credentials
49 fail the identity-echo probe.
50 """
51 if os.environ.get("GITHUB_ACTIONS", "").strip().casefold() != "true":
52 return
53 from .emulator import emulator_endpoint_requested, verify_emulator_endpoint
55 endpoint = emulator_endpoint_requested()
56 if endpoint is None:
57 raise RuntimeError(
58 "Live release validation is local-only and must not run in GitHub Actions"
59 )
60 verify_emulator_endpoint(endpoint)
63class LiveValidationRunner:
64 """Run selected live actions and always report and clean up."""
66 def __init__(
67 self,
68 settings: RunSettings,
69 registry: dict[str, ActionDefinition] | None = None,
70 ):
71 require_local_execution()
72 ensure_private_run_directory(settings.report_dir, settings.checkpoint_path)
73 self.settings = settings
74 # A sibling harness (scripts/example_job_validation) reuses this runner
75 # with its own action registry; the default remains the live release
76 # validation registry.
77 self.registry = registry if registry is not None else build_action_registry()
78 self._deploy_dependent_actions = self._derive_deploy_dependent_actions(self.registry)
79 self.selected_actions = self._resolve_actions(settings.requested_actions)
80 self._previous_cwd = Path.cwd()
81 os.chdir(settings.repo_root)
82 try:
83 self.cdk_context, self.deployment_regions = self._load_cdk_context(settings.repo_root)
84 self.config = self._build_config(self.cdk_context, self.deployment_regions)
85 self.checkpoint = self._load_checkpoint()
86 self.report = ValidationReport(
87 run_id=settings.run_id,
88 identity=settings.identity(),
89 selected_actions=list(self.selected_actions),
90 started_at=self.checkpoint.created_at,
91 action_results=list(self.checkpoint.action_results.values()),
92 baseline=self.checkpoint.baseline,
93 )
94 # Adaptive throttle retries for every harness client: the
95 # inventory scanners issue one metadata read per resource across
96 # every enabled Region, and a Regional TPS squeeze must surface
97 # as a bounded wait, not a failed action. See aws_session.py.
98 self.session = ThrottleResilientSession()
99 self.aws_client = GCOAWSClient(self.config)
100 self.aws_client._session = self.session
101 self.job_manager = JobManager(self.config)
102 self.job_manager._aws_client = self.aws_client
103 self.stack_manager = StackManager(self.config, project_root=settings.repo_root)
104 extra_cdk_context = settings.extra_cdk_context()
105 if extra_cdk_context:
106 # Force-enable the requested off-by-default features for
107 # every CDK invocation of this run (deploy, destroy, list all
108 # synthesize the same graph) without touching cdk.json — the
109 # preflight clean-worktree rule stays intact and the overrides
110 # are part of the checkpoint identity.
111 self.stack_manager.set_extra_cdk_context(extra_cdk_context)
112 self.context = RunContext(
113 settings=settings,
114 checkpoint=self.checkpoint,
115 report=self.report,
116 cdk_context=self.cdk_context,
117 deployment_regions=self.deployment_regions,
118 config=self.config,
119 session=self.session,
120 stack_manager=self.stack_manager,
121 aws_client=self.aws_client,
122 job_manager=self.job_manager,
123 persist_callback=self._persist_checkpoint,
124 )
125 except BaseException:
126 os.chdir(self._previous_cwd)
127 raise
128 self._identity_verified = False
129 self._received_signal: int | None = None
130 self._previous_signal_handlers: dict[int, Any] = {}
131 #: The guaranteed post-run inventory when it ran *after* the
132 #: ``final-inventory`` action had already passed: the action keeps its
133 #: own row and timing, and this scan (the latest look at the account)
134 #: becomes the report's final inventory.
135 self._final_inventory_recheck: ActionResult | None = None
137 def _install_signal_handlers(self) -> None:
138 """Route termination signals through the normal cleanup/report path."""
139 for signal_name in ("SIGTERM", "SIGHUP"):
140 signum = getattr(signal, signal_name, None)
141 if signum is None:
142 continue
143 self._previous_signal_handlers[signum] = signal.getsignal(signum)
144 signal.signal(signum, self._handle_signal)
146 def _restore_signal_handlers(self) -> None:
147 for signum, previous in self._previous_signal_handlers.items():
148 signal.signal(signum, previous)
149 self._previous_signal_handlers.clear()
151 def _handle_signal(self, signum: int, _frame: Any) -> None:
152 self._received_signal = signum
153 raise _LiveValidationSignal(signum)
155 @staticmethod
156 def _derive_deploy_dependent_actions(
157 registry: dict[str, ActionDefinition],
158 ) -> frozenset[str]:
159 """Actions that must not resume incomplete once teardown is recorded.
161 Derived from the registry rather than hardcoded: ``deploy`` itself plus
162 every action that transitively depends on it — except the teardown pair
163 (``destroy``/``final-inventory``), which exist precisely to run against
164 a destroyed deployment.
165 """
167 def depends_on_deploy(name: str, seen: frozenset[str] = frozenset()) -> bool:
168 if name == "deploy":
169 return True
170 if name in seen:
171 return False
172 return any(
173 depends_on_deploy(dep, seen | {name})
174 for dep in registry[name].dependencies
175 if dep in registry
176 )
178 return frozenset(
179 name
180 for name in registry
181 if name not in {"destroy", "final-inventory"} and depends_on_deploy(name)
182 )
184 @staticmethod
185 def _load_cdk_context(repo_root: Path) -> tuple[dict[str, Any], tuple[str, ...]]:
186 path = repo_root / "cdk.json"
187 try:
188 data = json.loads(path.read_text(encoding="utf-8"))
189 except (OSError, UnicodeError, json.JSONDecodeError) as exc:
190 raise ValueError(f"Unable to read {path}: {exc}") from exc
191 context = data.get("context") if isinstance(data, dict) else None
192 if not isinstance(context, dict):
193 raise ValueError("cdk.json context must be an object")
194 project_name = context.get("project_name")
195 regions = context.get("deployment_regions")
196 if not isinstance(project_name, str) or not project_name:
197 raise ValueError("cdk.json context.project_name must be a non-empty string")
198 if not isinstance(regions, dict):
199 raise ValueError("cdk.json context.deployment_regions must be an object")
200 for key in ("global", "api_gateway", "monitoring"):
201 if not isinstance(regions.get(key), str) or not regions[key]:
202 raise ValueError(f"cdk.json deployment_regions.{key} must be non-empty")
203 regional = regions.get("regional")
204 if (
205 not isinstance(regional, list)
206 or not regional
207 or any(not isinstance(item, str) or not item for item in regional)
208 ):
209 raise ValueError("cdk.json deployment_regions.regional must be a non-empty list")
210 if len(set(regional)) != len(regional):
211 raise ValueError("cdk.json deployment_regions.regional contains duplicates")
212 return context, tuple(regional)
214 @staticmethod
215 def _build_config(context: dict[str, Any], deployment_regions: tuple[str, ...]) -> GCOConfig:
216 regions = context["deployment_regions"]
217 return GCOConfig(
218 project_name=context["project_name"],
219 default_region=deployment_regions[0],
220 api_gateway_region=regions["api_gateway"],
221 global_region=regions["global"],
222 monitoring_region=regions["monitoring"],
223 default_namespace="gco-jobs",
224 output_format="json",
225 use_regional_api=False,
226 )
228 def _load_checkpoint(self) -> RunCheckpoint:
229 path = self.settings.checkpoint_path
230 if self.settings.resume:
231 if not path.is_file():
232 raise ValueError(f"--resume requires an existing checkpoint: {path}")
233 checkpoint = RunCheckpoint.from_path(path)
234 if checkpoint.identity != self.settings.identity():
235 raise ValueError(
236 "Checkpoint identity does not match this invocation. "
237 "Account, SHA, branch, profile, actions, run ID, repository, and "
238 "protected stacks must remain exact."
239 )
240 return checkpoint
241 if path.exists():
242 raise ValueError(
243 f"Checkpoint already exists: {path}. Use --resume with identical inputs "
244 "or choose a new --run-id/report directory."
245 )
246 checkpoint = RunCheckpoint(identity=self.settings.identity())
247 self._persist_checkpoint(checkpoint)
248 return checkpoint
250 def _persist_checkpoint(self, checkpoint: RunCheckpoint) -> None:
251 atomic_write_json(self.settings.checkpoint_path, checkpoint.to_dict())
253 def _resolve_actions(self, requested: tuple[str, ...]) -> tuple[str, ...]:
254 if not requested or "all" in requested:
255 if requested and len(requested) != 1:
256 raise ValueError("'all' cannot be combined with individual action names")
257 return tuple(self.registry)
259 unknown = sorted(set(requested) - set(self.registry))
260 if unknown:
261 raise ValueError(
262 f"Unknown actions: {', '.join(unknown)}. Available: " + ", ".join(self.registry)
263 )
265 selected: set[str] = set()
267 def include(name: str) -> None:
268 if name in selected:
269 return
270 for dependency in self.registry[name].dependencies:
271 include(dependency)
272 selected.add(name)
274 for name in requested:
275 include(name)
276 return tuple(name for name in self.registry if name in selected)
278 def _refresh_report_results(self) -> None:
279 ordered_names = list(self.registry)
280 self.report.action_results = [
281 self.checkpoint.action_results[name]
282 for name in ordered_names
283 if name in self.checkpoint.action_results
284 ]
285 self.report.baseline = self.checkpoint.baseline
286 final_result = self.checkpoint.action_results.get("final-inventory")
287 if self._final_inventory_recheck is not None:
288 # The post-run re-check is the most recent scan of the account.
289 self.report.final_inventory = self._final_inventory_recheck.details
290 elif final_result is not None and final_result.status == "passed":
291 self.report.final_inventory = final_result.details
293 def _write_report(self) -> tuple[Path, Path]:
294 self._refresh_report_results()
295 return self.report.write(self.settings.report_dir)
297 def _successful_status(self) -> Literal["passed", "partial"]:
298 """Reserve passed for execution of the complete action registry."""
299 if self.selected_actions == tuple(self.registry):
300 return "passed"
301 return "partial"
303 def _execute_action(
304 self,
305 definition: ActionDefinition,
306 *,
307 always_run: bool = False,
308 ) -> dict[str, Any]:
309 if (
310 not always_run
311 and definition.name != "preflight"
312 and definition.name in self.checkpoint.completed_actions
313 ):
314 result = self.checkpoint.action_results[definition.name]
315 print(f"[skip] {definition.name}: checkpoint already passed")
316 return result.details
318 if (
319 self.checkpoint.destroyed
320 and definition.name in self._deploy_dependent_actions
321 and definition.name not in self.checkpoint.completed_actions
322 ):
323 raise RuntimeError(
324 f"Cannot resume incomplete action {definition.name!r}: the checkpoint "
325 "already records infrastructure teardown"
326 )
328 print(f"[run] {definition.name}: {definition.description}")
329 started_at = utc_now()
330 started = time.monotonic()
331 try:
332 details = definition.handler(self.context)
333 except BaseException as exc:
334 result = ActionResult.failed(
335 name=definition.name,
336 description=definition.description,
337 started_at=started_at,
338 started_monotonic=started,
339 ended_monotonic=time.monotonic(),
340 error=exc,
341 )
342 self.checkpoint.action_results[definition.name] = result
343 if definition.name in self.checkpoint.completed_actions:
344 self.checkpoint.completed_actions.remove(definition.name)
345 self._persist_checkpoint(self.checkpoint)
346 self._write_report()
347 print(f"[fail] {definition.name}: {result.error}")
348 raise
350 result = ActionResult.passed(
351 name=definition.name,
352 description=definition.description,
353 started_at=started_at,
354 started_monotonic=started,
355 ended_monotonic=time.monotonic(),
356 details=details,
357 )
358 self.checkpoint.action_results[definition.name] = result
359 if definition.name not in self.checkpoint.completed_actions:
360 self.checkpoint.completed_actions.append(definition.name)
361 self._persist_checkpoint(self.checkpoint)
362 self._write_report()
363 print(f"[pass] {definition.name} ({result.duration_seconds:.3f}s)")
364 return details
366 def _guaranteed_cleanup(self) -> None:
367 if not self.checkpoint.deployment_attempted:
368 self.report.cleanup = {"needed": False}
369 return
370 if not self._identity_verified:
371 self.report.cleanup = {
372 "needed": True,
373 "completed": False,
374 "blocked": (
375 "Current invocation did not pass exact account/git preflight; "
376 "automatic cleanup was not allowed against an unverified identity"
377 ),
378 }
379 return
381 destroy_started_at = utc_now()
382 destroy_started = time.monotonic()
383 try:
384 details = destroy_deployment(self.context)
385 self.report.cleanup = {"completed": True, **details}
386 definition = self.registry["destroy"]
387 if "destroy" not in self.checkpoint.completed_actions:
388 # The action list never reached destroy (an earlier action
389 # failed, or the scope excluded it): this reconciliation is the
390 # run's teardown, so it is recorded as the destroy result with
391 # the time it actually took.
392 result = ActionResult.passed(
393 name="destroy",
394 description=definition.description,
395 started_at=destroy_started_at,
396 started_monotonic=destroy_started,
397 ended_monotonic=time.monotonic(),
398 details=details,
399 )
400 self.checkpoint.action_results["destroy"] = result
401 self.checkpoint.completed_actions.append("destroy")
402 self._persist_checkpoint(self.checkpoint)
403 except _LiveValidationSignal, KeyboardInterrupt:
404 raise
405 except BaseException as exc:
406 self.report.cleanup = {
407 "needed": True,
408 "completed": False,
409 "error": f"{type(exc).__name__}: {exc}",
410 "traceback": "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
411 "workload_cleanup_attempts": self.checkpoint.state.get(
412 "workload_cleanup_attempts", []
413 ),
414 "attempts": self.checkpoint.state.get("destroy_attempts", []),
415 "retained_cleanup_attempts": self.checkpoint.state.get(
416 "retained_cleanup_attempts", []
417 ),
418 }
420 if self.checkpoint.baseline is not None:
421 self._recheck_final_inventory()
423 def _recheck_final_inventory(self) -> None:
424 """Run the final inventory once more after the guaranteed teardown.
426 This is the last look at the account, so it always runs when a baseline
427 exists. What it records depends on what the action list already did:
429 * If the ``final-inventory`` action never passed (it was skipped after an
430 earlier failure, or excluded from the scope), this scan *is* the run's
431 final inventory and becomes the action result, with the time it took.
432 * If the action already passed, its row and duration stand; the re-check
433 is recorded under ``cleanup.final_inventory_recheck`` and its details
434 become the report's final inventory, being the most recent scan.
435 * A failed re-check is authoritative either way: the account is not
436 clean, so it replaces any passed result and drops the completion.
438 Before this method existed the re-check overwrote the action's row with a
439 zero-duration result, so every report claimed its 10-minute inventory took
440 0.000s.
441 """
442 definition = self.registry["final-inventory"]
443 started_at = utc_now()
444 started = time.monotonic()
445 try:
446 details = action_final_inventory(self.context)
447 except _LiveValidationSignal, KeyboardInterrupt:
448 raise
449 # BaseException on purpose, matching the other cleanup paths in this file:
450 # the interrupts are re-raised above, so this leaves SystemExit from CLI
451 # helpers under the scan. Recording it here keeps it a failed inventory
452 # instead of run()'s outer handler marking a finished teardown incomplete.
453 except BaseException as exc:
454 self.checkpoint.action_results["final-inventory"] = ActionResult.failed(
455 name="final-inventory",
456 description=definition.description,
457 started_at=started_at,
458 started_monotonic=started,
459 ended_monotonic=time.monotonic(),
460 error=exc,
461 )
462 if "final-inventory" in self.checkpoint.completed_actions:
463 self.checkpoint.completed_actions.remove("final-inventory")
464 self._persist_checkpoint(self.checkpoint)
465 return
467 recheck = ActionResult.passed(
468 name="final-inventory",
469 description=definition.description,
470 started_at=started_at,
471 started_monotonic=started,
472 ended_monotonic=time.monotonic(),
473 details=details,
474 )
475 previous = self.checkpoint.action_results.get("final-inventory")
476 supersedes = previous is None or previous.status != "passed"
477 if supersedes:
478 self.checkpoint.action_results["final-inventory"] = recheck
479 else:
480 self._final_inventory_recheck = recheck
481 self.report.cleanup["final_inventory_recheck"] = {
482 "status": recheck.status,
483 "started_at": recheck.started_at,
484 "ended_at": recheck.ended_at,
485 "duration_seconds": recheck.duration_seconds,
486 "recorded_as_action_result": supersedes,
487 }
488 if "final-inventory" not in self.checkpoint.completed_actions:
489 self.checkpoint.completed_actions.append("final-inventory")
490 self._persist_checkpoint(self.checkpoint)
492 def run(self) -> int:
493 """Execute selected actions, then report and clean up in all cases."""
494 failure: BaseException | None = None
495 interrupted = False
496 interrupt_exit_code: int | None = None
497 try:
498 self._install_signal_handlers()
499 try:
500 for name in self.selected_actions:
501 definition = self.registry[name]
502 self._execute_action(definition)
503 if name == "preflight":
504 self._identity_verified = True
505 except _LiveValidationSignal as exc:
506 interrupted = True
507 interrupt_exit_code = 128 + exc.signum
508 failure = exc
509 self.report.fatal_error = (
510 f"{exc.signal_name}: validation interrupted; controlled cleanup started"
511 )
512 except KeyboardInterrupt as exc:
513 interrupted = True
514 interrupt_exit_code = 130
515 failure = exc
516 self.report.fatal_error = "KeyboardInterrupt: validation interrupted"
517 except BaseException as exc:
518 failure = exc
519 self.report.fatal_error = "".join(
520 traceback.format_exception(type(exc), exc, exc.__traceback__)
521 )
522 finally:
523 try:
524 if self.checkpoint.deployment_attempted:
525 # Always reconcile exact target-stack absence on resume, even
526 # when an older checkpoint says teardown completed. This lets
527 # destroy_deployment reopen stale terminal state safely.
528 self._guaranteed_cleanup()
529 else:
530 self.report.cleanup = {"needed": False}
531 except BaseException as cleanup_exc:
532 if isinstance(cleanup_exc, _LiveValidationSignal):
533 interrupted = True
534 interrupt_exit_code = 128 + cleanup_exc.signum
535 failure = cleanup_exc
536 self.report.fatal_error = (
537 f"{cleanup_exc.signal_name}: validation interrupted during cleanup"
538 )
539 elif isinstance(cleanup_exc, KeyboardInterrupt):
540 interrupted = True
541 interrupt_exit_code = 130
542 failure = cleanup_exc
543 self.report.fatal_error = "KeyboardInterrupt: cleanup interrupted"
544 self.report.cleanup = {
545 "needed": True,
546 "completed": False,
547 "runner_error": f"{type(cleanup_exc).__name__}: {cleanup_exc}",
548 "workload_cleanup_attempts": self.checkpoint.state.get(
549 "workload_cleanup_attempts", []
550 ),
551 "attempts": self.checkpoint.state.get("destroy_attempts", []),
552 "retained_cleanup_attempts": self.checkpoint.state.get(
553 "retained_cleanup_attempts", []
554 ),
555 }
557 self._refresh_report_results()
558 failed_results = [
559 result for result in self.report.action_results if result.status == "failed"
560 ]
561 cleanup_failed = bool(
562 self.checkpoint.deployment_attempted
563 and not self.report.cleanup.get("completed", False)
564 )
565 if interrupted:
566 self.report.status = "interrupted"
567 elif failure is not None or failed_results or cleanup_failed:
568 self.report.status = "failed"
569 else:
570 self.report.status = self._successful_status()
571 self.report.ended_at = utc_now()
572 json_path, markdown_path = self._write_report()
573 print(f"JSON report: {json_path}")
574 print(f"Markdown report: {markdown_path}")
575 finally:
576 self._restore_signal_handlers()
577 os.chdir(self._previous_cwd)
579 if interrupt_exit_code is not None:
580 return interrupt_exit_code
581 return 0 if self.report.status in {"passed", "partial"} else 1