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