Coverage for scripts / live_release_validation / checks / jobs.py: 100.00%
318 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Job submission, appearance, terminal-state, and deletion helpers."""
3from __future__ import annotations
5import copy
6import hashlib
7import re
8import time
9from typing import Any, cast
10from urllib.parse import quote
12from cli.jobs import resolve_submission_identity
14from ..constants import (
15 _CENTRAL_MANAGED_BY_LABEL,
16 _CENTRAL_ORIGINAL_NAME_ANNOTATION,
17 _CENTRAL_QUEUE_ID_ANNOTATION,
18 _CENTRAL_QUEUE_KEY_LABEL,
19 _MANIFEST_DIR,
20 _PATH_JOB_LABEL,
21 _RUN_JOB_LABEL,
22)
23from ..context import (
24 _job_transport_region,
25)
26from ..models import RunContext
29def _run_token(run_id: str) -> str:
30 token = re.sub(r"[^a-z0-9-]+", "-", run_id.lower()).strip("-")
31 token = re.sub(r"-+", "-", token)[:24].rstrip("-")
32 if not token:
33 raise RuntimeError("run_id does not contain a Kubernetes-safe token")
34 return token
37def _replace_token(value: Any, token: str) -> Any:
38 if isinstance(value, str):
39 return value.replace("__RUN_TOKEN__", token)
40 if isinstance(value, list):
41 return [_replace_token(item, token) for item in value]
42 if isinstance(value, dict):
43 return {key: _replace_token(item, token) for key, item in value.items()}
44 return value
47def _load_manifest(ctx: RunContext, filename: str) -> tuple[list[dict[str, Any]], str, str]:
48 # _MANIFEST_DIR is anchored at the package root by constants.py; never
49 # resolve manifests relative to this module's __file__ (that is exactly
50 # what failed in run retry1-8002d6c80f62 when this helper moved here).
51 path = _MANIFEST_DIR / filename
52 manifests = ctx.job_manager.load_manifests(str(path))
53 manifests = _replace_token(manifests, _run_token(ctx.settings.run_id))
54 job = next(item for item in manifests if item.get("kind") == "Job")
55 name = str(job["metadata"]["name"])
56 namespace = str(job["metadata"]["namespace"])
57 return manifests, name, namespace
60def _central_workload_identity(record: dict[str, Any]) -> tuple[str, str, str] | None:
61 values = (
62 record.get("k8s_job_name"),
63 record.get("k8s_job_namespace"),
64 record.get("k8s_job_uid"),
65 )
66 populated = [value is not None for value in values]
67 if any(populated) and not all(populated):
68 raise RuntimeError("Checkpoint contains a partial central Kubernetes identity")
69 if not any(populated):
70 return None
71 identity = tuple(str(value or "") for value in values)
72 if not all(identity):
73 raise RuntimeError("Checkpoint contains an empty central Kubernetes identity field")
74 return cast(tuple[str, str, str], identity)
77def _effective_job_identity(record: dict[str, Any]) -> tuple[str, str]:
78 if record.get("path") == "dynamodb":
79 central_identity = _central_workload_identity(record)
80 if central_identity is None:
81 raise RuntimeError(
82 "Central workload Kubernetes identity has not been bound from DynamoDB"
83 )
84 return central_identity[0], central_identity[1]
85 return str(record["name"]), str(record["namespace"])
88def _job_reference_identity(record: dict[str, Any]) -> tuple[str, str]:
89 if record.get("path") == "dynamodb":
90 central_identity = _central_workload_identity(record)
91 if central_identity is not None:
92 return central_identity[0], central_identity[1]
93 return str(record["name"]), str(record["namespace"])
96def _job_api_path(record: dict[str, Any], suffix: str = "") -> str:
97 actual_name, actual_namespace = _effective_job_identity(record)
98 namespace = quote(actual_namespace, safe="")
99 name = quote(actual_name, safe="")
100 return f"/api/v1/jobs/{namespace}/{name}{suffix}"
103def _response_json(response: Any, operation: str) -> dict[str, Any]:
104 try:
105 value = response.json()
106 except (TypeError, ValueError) as exc:
107 raise RuntimeError(f"{operation} returned invalid JSON: {response.text}") from exc
108 if not isinstance(value, dict):
109 raise RuntimeError(f"{operation} returned a non-object JSON response")
110 return value
113def _verify_response_region(data: dict[str, Any], expected_region: str, operation: str) -> None:
114 actual_region = str(data.get("region") or "")
115 if actual_region != expected_region:
116 raise RuntimeError(
117 f"{operation} came from Region {actual_region or 'unknown'}, expected {expected_region}"
118 )
121def _validate_central_workload_metadata(
122 record: dict[str, Any],
123 metadata: dict[str, Any],
124 labels: dict[str, Any],
125 uid: str,
126) -> None:
127 queue_job_id = str(record.get("central_queue_job_id") or "")
128 expected_uid = str(record.get("k8s_job_uid") or "")
129 if not queue_job_id or not expected_uid:
130 raise RuntimeError("Central workload is missing immutable queue/UID authority")
131 if uid != expected_uid:
132 raise RuntimeError("Kubernetes Job UID differs from persisted central worker identity")
133 if labels.get(_CENTRAL_MANAGED_BY_LABEL) != "central-queue":
134 raise RuntimeError("Central Job managed-by label does not match the worker contract")
135 expected_queue_key = hashlib.sha256(queue_job_id.encode("utf-8")).hexdigest()[:32]
136 if labels.get(_CENTRAL_QUEUE_KEY_LABEL) != expected_queue_key:
137 raise RuntimeError("Central Job queue-key label does not match its queue ID")
138 annotations = metadata.get("annotations")
139 if not isinstance(annotations, dict):
140 raise RuntimeError("Central Job lookup omitted ownership annotations")
141 if annotations.get(_CENTRAL_QUEUE_ID_ANNOTATION) != queue_job_id:
142 raise RuntimeError("Central Job queue ID annotation does not match the checkpoint")
143 if annotations.get(_CENTRAL_ORIGINAL_NAME_ANNOTATION) != record["name"]:
144 raise RuntimeError("Central Job original-name annotation does not match the request")
147def _get_owned_job(ctx: RunContext, record: dict[str, Any]) -> dict[str, Any] | None:
148 """Return a Job only after authoritative HTTP and UID/label verification."""
149 actual_name, actual_namespace = _effective_job_identity(record)
150 response = ctx.aws_client.make_authenticated_request(
151 method="GET",
152 path=_job_api_path(record),
153 target_region=record.get("transport_region"),
154 )
155 if response.status_code == 404:
156 return None
157 if not response.ok:
158 raise RuntimeError(
159 f"Job lookup failed for {record['region']}:{actual_namespace}/{actual_name}: "
160 f"{response.status_code} {response.text}"
161 )
162 data = _response_json(response, "Job lookup")
163 _verify_response_region(data, str(record["region"]), "Job lookup")
164 metadata = data.get("metadata")
165 if not isinstance(metadata, dict):
166 raise RuntimeError("Job lookup omitted metadata")
167 if metadata.get("name") != actual_name or metadata.get("namespace") != actual_namespace:
168 raise RuntimeError("Job lookup returned a different Kubernetes identity")
169 labels = metadata.get("labels")
170 if not isinstance(labels, dict):
171 raise RuntimeError("Job lookup omitted ownership labels")
172 if labels.get(_RUN_JOB_LABEL) != record["run_label"]:
173 raise RuntimeError("Job run label does not match the checkpoint")
174 if labels.get(_PATH_JOB_LABEL) != record["path"]:
175 raise RuntimeError("Job validation-path label does not match the checkpoint")
176 uid = str(metadata.get("uid") or "")
177 if not uid:
178 raise RuntimeError("Job lookup omitted metadata.uid")
179 if record.get("path") == "dynamodb":
180 _validate_central_workload_metadata(record, metadata, labels, uid)
181 ctx.record_job_uid(record, uid)
182 return data
185def _reactivate_deleted_job_record(ctx: RunContext, record: dict[str, Any]) -> None:
186 """Permit a crash-window replay only after authoritative prior absence."""
187 if not record.get("deleted"):
188 return
189 if _get_owned_job(ctx, record) is not None:
190 raise RuntimeError("Checkpoint marks a Job deleted but the exact UID still exists")
191 with ctx.state_lock:
192 previous_uid = record.get("uid")
193 if previous_uid:
194 record.setdefault("previous_uids", []).append(previous_uid)
195 record["uid"] = None
196 record["deleted"] = False
197 record["submission_state"] = "registered"
198 for key in (
199 "submission_started_at",
200 "submission_reconcile_deadline",
201 "submission_acknowledged_at",
202 "appearance_deadline",
203 "submission",
204 "submission_envelope",
205 "submission_resumable",
206 "submission_blocked_reason",
207 "submission_blocked_at",
208 "not_submitted_at",
209 "validation_evidence",
210 "deleted_at",
211 ):
212 record.pop(key, None)
213 ctx.persist_callback(ctx.checkpoint)
216def _job_appearance_timeout(ctx: RunContext) -> int:
217 return min(ctx.settings.job_timeout_seconds, ctx.settings.queue_timeout_seconds)
220def _wait_for_owned_job_appearance(
221 ctx: RunContext,
222 record: dict[str, Any],
223 *,
224 raise_on_timeout: bool = True,
225) -> dict[str, Any] | None:
226 raw_deadline = record.get("appearance_deadline")
227 if raw_deadline is None:
228 deadline = time.time() + _job_appearance_timeout(ctx)
229 with ctx.state_lock:
230 record["appearance_deadline"] = deadline
231 ctx.persist_callback(ctx.checkpoint)
232 else:
233 deadline = float(raw_deadline)
234 while True:
235 job = _get_owned_job(ctx, record)
236 if job is not None:
237 return job
238 if time.time() >= deadline:
239 if raise_on_timeout:
240 actual_name, actual_namespace = _effective_job_identity(record)
241 raise TimeoutError(
242 f"Job {record['region']}:{actual_namespace}/{actual_name} "
243 "did not appear before the bounded submission deadline"
244 )
245 return None
246 time.sleep(ctx.settings.poll_interval_seconds)
249def _wait_for_ambiguous_job_reconciliation(
250 ctx: RunContext,
251 record: dict[str, Any],
252) -> dict[str, Any] | None:
253 """Observe a non-replayable escaped submission until its distinct deadline."""
254 raw_deadline = record.get("submission_reconcile_deadline")
255 if raw_deadline is None:
256 raise RuntimeError("Ambiguous Job submission has no reconciliation deadline")
257 deadline = float(raw_deadline)
258 while True:
259 job = _get_owned_job(ctx, record)
260 if job is not None:
261 return job
262 if time.time() >= deadline:
263 return None
264 time.sleep(ctx.settings.poll_interval_seconds)
267def _job_status(job: dict[str, Any]) -> str:
268 status = job.get("status") or {}
269 for condition in status.get("conditions") or []:
270 if condition.get("type") == "Complete" and condition.get("status") == "True":
271 return "succeeded"
272 if condition.get("type") == "Failed" and condition.get("status") == "True":
273 return "failed"
274 return "running" if int(status.get("active") or 0) > 0 else "pending"
277def _wait_for_owned_job_terminal(
278 ctx: RunContext, record: dict[str, Any]
279) -> tuple[dict[str, Any], list[dict[str, Any]]]:
280 deadline = time.monotonic() + ctx.settings.job_timeout_seconds
281 history: list[dict[str, Any]] = []
282 while True:
283 job = _get_owned_job(ctx, record)
284 if job is None:
285 raise RuntimeError("An owned Job disappeared before reaching a terminal state")
286 status = _job_status(job)
287 history.append({"at": time.time(), "status": status})
288 if status in {"succeeded", "failed"}:
289 return job, history
290 if time.monotonic() >= deadline:
291 actual_name, actual_namespace = _effective_job_identity(record)
292 raise TimeoutError(
293 f"Job {record['region']}:{actual_namespace}/{actual_name} "
294 f"did not complete within {ctx.settings.job_timeout_seconds}s"
295 )
296 time.sleep(ctx.settings.poll_interval_seconds)
299def _owned_job_logs(ctx: RunContext, record: dict[str, Any], tail: int = 200) -> str:
300 if _get_owned_job(ctx, record) is None:
301 raise RuntimeError("Owned Job disappeared before its logs were read")
302 actual_name, actual_namespace = _effective_job_identity(record)
303 response = ctx.aws_client.make_authenticated_request(
304 method="GET",
305 path=_job_api_path(record, f"/logs?tail={tail}"),
306 target_region=record.get("transport_region"),
307 )
308 if not response.ok:
309 raise RuntimeError(f"Job log lookup failed: {response.status_code} {response.text}")
310 data = _response_json(response, "Job log lookup")
311 _verify_response_region(data, str(record["region"]), "Job log lookup")
312 if data.get("job_name") != actual_name or data.get("namespace") != actual_namespace:
313 raise RuntimeError("Job log lookup returned a different Job identity")
314 return str(data.get("logs") or "")
317def _wait_for_owned_job_absence(ctx: RunContext, record: dict[str, Any]) -> None:
318 consecutive_absent = 0
319 deadline = time.monotonic() + 180
320 while True:
321 current = _get_owned_job(ctx, record)
322 if current is None:
323 consecutive_absent += 1
324 if consecutive_absent >= 3:
325 return
326 else:
327 consecutive_absent = 0
328 if time.monotonic() >= deadline:
329 actual_name, actual_namespace = _effective_job_identity(record)
330 raise TimeoutError(
331 f"Job {record['region']}:{actual_namespace}/{actual_name} remained visible"
332 )
333 time.sleep(min(5, ctx.settings.poll_interval_seconds))
336def _delete_owned_job(ctx: RunContext, record: dict[str, Any]) -> dict[str, Any]:
337 state = str(record.get("submission_state") or "registered")
338 if record.get("path") == "dynamodb" and _central_workload_identity(record) is None:
339 if state in {"registered", "prepared", "not_submitted"}:
340 ctx.mark_job_not_submitted(record)
341 ctx.mark_job_deleted(record)
342 return {"not_submitted": True, "already_absent": True}
343 raise RuntimeError(
344 "Central Job submission may have escaped but no worker-persisted Kubernetes "
345 "identity was bound; cleanup remains unresolved"
346 )
348 current = _get_owned_job(ctx, record)
349 if current is None and state == "submitting" and not record.get("uid"):
350 current = _wait_for_ambiguous_job_reconciliation(ctx, record)
351 elif current is None and state == "submitted" and not record.get("uid"):
352 current = _wait_for_owned_job_appearance(ctx, record, raise_on_timeout=False)
353 if current is None:
354 if record.get("uid"):
355 _wait_for_owned_job_absence(ctx, record)
356 ctx.mark_job_deleted(record)
357 return {"authoritative_absence_after_uid_observation": True}
358 if state in {"registered", "prepared", "not_submitted"}:
359 ctx.mark_job_not_submitted(record)
360 ctx.mark_job_deleted(record)
361 return {"not_submitted": True, "already_absent": True}
362 raise RuntimeError(
363 "Job submission may have escaped but no immutable Kubernetes UID was observed; "
364 "cleanup remains unresolved"
365 )
367 expected_uid = str(record.get("uid") or "")
368 if not expected_uid:
369 raise RuntimeError("Owned Job has no checkpointed UID at deletion time")
370 separator = "&" if "?" in _job_api_path(record) else "?"
371 response = ctx.aws_client.make_authenticated_request(
372 method="DELETE",
373 path=(f"{_job_api_path(record)}{separator}expected_uid={quote(expected_uid, safe='')}"),
374 target_region=record.get("transport_region"),
375 )
376 if response.status_code == 404:
377 deletion: dict[str, Any] = {"authoritative_404_after_uid_observation": True}
378 elif response.status_code == 409:
379 raise RuntimeError("Job UID changed before deletion; Kubernetes precondition rejected it")
380 elif response.ok:
381 deletion = _response_json(response, "Job deletion")
382 _verify_response_region(deletion, str(record["region"]), "Job deletion")
383 response_uid = deletion.get("uid")
384 if response_uid is not None and str(response_uid) != expected_uid:
385 raise RuntimeError("Job deletion response UID did not match the checkpoint")
386 else:
387 raise RuntimeError(f"Job deletion failed: {response.status_code} {response.text}")
388 _wait_for_owned_job_absence(ctx, record)
389 ctx.mark_job_deleted(record)
390 return deletion
393def _complete_job_lifecycle(
394 ctx: RunContext,
395 *,
396 record: dict[str, Any],
397 marker: str,
398) -> dict[str, Any]:
399 appeared = _wait_for_owned_job_appearance(ctx, record)
400 actual_name, actual_namespace = _effective_job_identity(record)
401 if appeared is None:
402 raise RuntimeError(
403 f"Job {actual_namespace}/{actual_name} never appeared in {record['region']}"
404 )
405 final, history = _wait_for_owned_job_terminal(ctx, record)
406 status = _job_status(final)
407 if status != "succeeded":
408 raise RuntimeError(
409 f"Job {actual_namespace}/{actual_name} in {record['region']} "
410 f"finished with status {status}"
411 )
412 logs = _owned_job_logs(ctx, record)
413 if marker not in logs:
414 raise RuntimeError(f"Job logs did not contain expected marker {marker!r}")
415 evidence = {
416 "name": actual_name,
417 "namespace": actual_namespace,
418 "requested_name": record["name"],
419 "requested_namespace": record["namespace"],
420 "region": record["region"],
421 "transport_region": record.get("transport_region"),
422 "uid": record.get("uid"),
423 "central_queue_job_id": record.get("central_queue_job_id"),
424 "status": status,
425 "status_history": history,
426 "marker": marker,
427 "appearance": {
428 "region": appeared.get("region"),
429 "uid": (appeared.get("metadata") or {}).get("uid"),
430 },
431 }
432 with ctx.state_lock:
433 record["validation_evidence"] = copy.deepcopy(evidence)
434 ctx.persist_callback(ctx.checkpoint)
435 deletion = _delete_owned_job(ctx, record)
436 return {**evidence, "deletion": deletion}
439def _register_job(
440 ctx: RunContext,
441 *,
442 name: str,
443 namespace: str,
444 execution_region: str,
445 path: str,
446 reactivate_deleted: bool = True,
447) -> dict[str, Any]:
448 record = ctx.register_job(
449 name=name,
450 namespace=namespace,
451 region=execution_region,
452 path=path,
453 run_label=_run_token(ctx.settings.run_id),
454 transport_region=_job_transport_region(ctx, execution_region),
455 )
456 if reactivate_deleted:
457 _reactivate_deleted_job_record(ctx, record)
458 return record
461def _run_api_transport_lifecycle(
462 ctx: RunContext,
463 *,
464 manifest_filename: str,
465 path: str,
466 marker_prefix: str,
467) -> dict[str, Any]:
468 """Run one manifest's complete authenticated-API Job lifecycle.
470 The crash-safe submission dance shared by the ``api`` action and every
471 scheduler probe: register the deterministic record, persist the envelope,
472 reconcile any escaped prior submission, submit through the manifest API,
473 then observe appearance, completion, the log marker, and deletion. The
474 marker is ``GCO_LIVE_<MARKER_PREFIX>_<run token>`` and must be emitted by
475 the manifest's workload.
476 """
477 manifests, name, namespace = _load_manifest(ctx, manifest_filename)
478 token = _run_token(ctx.settings.run_id)
479 marker = f"GCO_LIVE_{marker_prefix}_{token}"
480 execution_region = ctx.deployment_regions[0]
481 record = _register_job(
482 ctx,
483 name=name,
484 namespace=namespace,
485 execution_region=execution_region,
486 path=path,
487 )
488 envelope = {
489 "transport": "api",
490 "manifests": manifests,
491 "namespace": namespace,
492 "execution_region": execution_region,
493 "transport_region": record.get("transport_region"),
494 "labels": {_RUN_JOB_LABEL: token},
495 }
496 ctx.prepare_job_submission(record, envelope=envelope, resumable=False)
498 existing = _get_owned_job(ctx, record)
499 submission: dict[str, Any] | None = None
500 state = str(record.get("submission_state") or "")
501 if existing is None and state == "submitting":
502 existing = _wait_for_ambiguous_job_reconciliation(ctx, record)
503 if existing is None:
504 reason = (
505 f"{path} submission crossed a non-idempotent boundary but no Job "
506 "appeared; automatic replay is forbidden"
507 )
508 ctx.block_job_submission(record, reason)
509 raise RuntimeError(reason)
510 elif existing is None and state == "submitted":
511 existing = _wait_for_owned_job_appearance(ctx, record)
512 elif existing is None and state == "blocked":
513 raise RuntimeError(
514 str(record.get("submission_blocked_reason") or f"{path} submission blocked")
515 )
517 if existing is None:
518 if state != "prepared":
519 raise RuntimeError(f"Cannot submit {path} Job from state {state!r}")
520 ctx.begin_job_submission(
521 record,
522 reconciliation_timeout_seconds=_job_appearance_timeout(ctx),
523 )
524 submission = ctx.job_manager.submit_job(
525 manifests,
526 namespace=namespace,
527 target_region=record.get("transport_region"),
528 labels={_RUN_JOB_LABEL: token},
529 )
530 submitted_name, submitted_namespace = resolve_submission_identity(
531 submission,
532 fallback_name=name,
533 fallback_namespace=namespace,
534 )
535 if submitted_name != name or submitted_namespace != namespace:
536 raise RuntimeError(
537 f"{path} submission identity mismatch: "
538 f"expected {namespace}/{name}, got {submitted_namespace}/{submitted_name}"
539 )
540 response_region = submission.get("region")
541 if response_region is not None and str(response_region) != execution_region:
542 raise RuntimeError(
543 f"{path} submission executed in {response_region}, expected {execution_region}"
544 )
545 ctx.finish_job_submission(
546 record,
547 submission,
548 appearance_timeout_seconds=_job_appearance_timeout(ctx),
549 )
551 lifecycle = _complete_job_lifecycle(ctx, record=record, marker=marker)
552 lifecycle["submission"] = submission or {"reconciled_existing_job": True}
553 return lifecycle