Coverage for scripts / live_release_validation / cleanup / workloads.py: 100.00%
108 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Delete exactly run-owned Kubernetes workloads and queue records."""
3from __future__ import annotations
5import copy
6import time
7from typing import Any
8from urllib.parse import quote
10from ..checks.central_queue import (
11 _central_workload_record,
12 _read_central_job_item,
13 _reconcile_central_cleanup_workload,
14 _validate_central_job_identity,
15 _wait_for_central_queue_appearance,
16 _wait_for_central_queue_terminal,
17)
18from ..checks.jobs import (
19 _delete_owned_job,
20 _job_reference_identity,
21 _response_json,
22)
23from ..constants import (
24 _TERMINAL_QUEUE_STATUSES,
25)
26from ..models import RunContext, utc_now
29def _cleanup_central_job(ctx: RunContext, central_job: dict[str, Any]) -> dict[str, Any]:
30 job_id = str(central_job["job_id"])
31 current = _wait_for_central_queue_appearance(
32 ctx,
33 central_job,
34 raise_on_timeout=False,
35 )
36 if current is None:
37 outcome = {
38 "job_id": job_id,
39 "complete": False,
40 "unresolved": "no consistently readable queue record before the bounded deadline",
41 }
42 central_job["cleanup_complete"] = False
43 central_job["cleanup_result"] = outcome
44 ctx.persist()
45 raise RuntimeError(
46 f"Central queue job {job_id} was not observed; non-observation is not terminal proof"
47 )
49 status = str(current.get("status") or "unknown")
50 previous_cancellation = central_job.get("cancellation")
51 cancellation: dict[str, Any] = (
52 copy.deepcopy(previous_cancellation)
53 if isinstance(previous_cancellation, dict)
54 else {"not_required": status in _TERMINAL_QUEUE_STATUSES}
55 )
56 if status not in _TERMINAL_QUEUE_STATUSES:
57 reason = quote("live release validation cleanup", safe="")
58 response = ctx.aws_client.make_authenticated_request(
59 method="DELETE",
60 path=f"/api/v1/queue/jobs/{quote(job_id, safe='')}?reason={reason}",
61 target_region=central_job.get("transport_region"),
62 )
63 if response.status_code == 404:
64 raise RuntimeError(f"Central queue job {job_id} disappeared during cancellation")
65 if response.status_code == 409:
66 cancellation = {
67 "not_cancellable": True,
68 "status_code": 409,
69 "detail": response.text,
70 }
71 elif response.ok:
72 cancellation = {
73 "accepted_before_claim": True,
74 "response": _response_json(response, "Central queue cancellation"),
75 }
76 else:
77 raise RuntimeError(f"{response.status_code} {response.text}")
78 central_job["cancel_attempted"] = True
79 central_job["cancellation"] = cancellation
80 ctx.persist()
81 current, history = _wait_for_central_queue_terminal(ctx, central_job)
82 else:
83 history = [{"status": status, "at": time.time()}]
85 terminal_status = str(current.get("status") or "unknown")
86 if terminal_status not in _TERMINAL_QUEUE_STATUSES:
87 raise RuntimeError(f"Central queue job {job_id} did not become terminal")
88 persisted = _read_central_job_item(ctx, job_id)
89 _validate_central_job_identity(central_job, persisted)
90 persisted_status = str(persisted.get("status") or "unknown")
91 if persisted_status != terminal_status or persisted_status not in _TERMINAL_QUEUE_STATUSES:
92 raise RuntimeError(
93 f"Central queue job {job_id} lacks consistent terminal DynamoDB evidence"
94 )
96 _, workload_not_submitted = _reconcile_central_cleanup_workload(
97 ctx,
98 central_job,
99 persisted,
100 )
102 outcome = {
103 "job_id": job_id,
104 "complete": True,
105 "cancellation": cancellation,
106 "terminal_status": terminal_status,
107 "status_history": history,
108 "consistent_record": persisted,
109 "workload_not_submitted": workload_not_submitted,
110 }
111 central_job["status"] = terminal_status
112 central_job["cleanup_complete"] = True
113 central_job["cleanup_result"] = outcome
114 ctx.persist()
115 return outcome
118def cleanup_workloads(ctx: RunContext) -> dict[str, Any]:
119 """Reconcile every workload and return an explicit teardown barrier result."""
120 result: dict[str, Any] = {
121 "started_at": utc_now(),
122 "complete": False,
123 "jobs": [],
124 "central_jobs": [],
125 "errors": [],
126 "unresolved": [],
127 }
128 reconciled_central_workloads: set[int] = set()
129 for central_job in ctx.checkpoint.state.get("central_jobs", []):
130 job_id = str(central_job["job_id"])
131 try:
132 if central_job.get("cleanup_complete"):
133 persisted = _read_central_job_item(ctx, job_id)
134 _validate_central_job_identity(central_job, persisted)
135 persisted_status = str(persisted.get("status") or "unknown")
136 checkpoint_status = str(
137 central_job.get("status")
138 or (central_job.get("cleanup_result") or {}).get("terminal_status")
139 or ""
140 )
141 if persisted_status not in _TERMINAL_QUEUE_STATUSES:
142 raise RuntimeError(
143 f"Previously completed central cleanup for {job_id} is no longer terminal"
144 )
145 if checkpoint_status and checkpoint_status != persisted_status:
146 raise RuntimeError(
147 f"Central cleanup status changed from {checkpoint_status} "
148 f"to {persisted_status}"
149 )
150 workload_record, _ = _reconcile_central_cleanup_workload(
151 ctx,
152 central_job,
153 persisted,
154 )
155 result["central_jobs"].append(
156 copy.deepcopy(central_job.get("cleanup_result") or {})
157 )
158 else:
159 result["central_jobs"].append(_cleanup_central_job(ctx, central_job))
160 workload_record = _central_workload_record(ctx, central_job)
161 reconciled_central_workloads.add(id(workload_record))
162 except Exception as exc: # noqa: BLE001 - preserve every unresolved resource
163 error = f"{type(exc).__name__}: {exc}"
164 result["errors"].append({"resource": f"central:{job_id}", "error": error})
165 result["unresolved"].append({"resource": f"central:{job_id}", "reason": error})
167 for record in ctx.checkpoint.state.get("jobs", []):
168 if record.get("deleted"):
169 continue
170 requested_reference = f"{record['region']}:{record['namespace']}/{record['name']}"
171 reference = requested_reference
172 try:
173 if record.get("path") == "dynamodb" and id(record) not in reconciled_central_workloads:
174 raise RuntimeError(
175 "Central workload was not reconciled from terminal DynamoDB evidence "
176 "in this cleanup attempt"
177 )
178 actual_name, actual_namespace = _job_reference_identity(record)
179 reference = f"{record['region']}:{actual_namespace}/{actual_name}"
180 deletion = _delete_owned_job(ctx, record)
181 result["jobs"].append(
182 {
183 "region": record["region"],
184 "namespace": actual_namespace,
185 "name": actual_name,
186 "requested_namespace": record["namespace"],
187 "requested_name": record["name"],
188 "uid": record.get("uid"),
189 "deletion": deletion,
190 }
191 )
192 except Exception as exc: # noqa: BLE001 - preserve every unresolved resource
193 error = f"{type(exc).__name__}: {exc}"
194 result["errors"].append({"resource": reference, "error": error})
195 result["unresolved"].append({"resource": reference, "reason": error})
197 for central_job in ctx.checkpoint.state.get("central_jobs", []):
198 if not central_job.get("cleanup_complete"):
199 reference = f"central:{central_job['job_id']}"
200 if not any(item["resource"] == reference for item in result["unresolved"]):
201 result["unresolved"].append(
202 {"resource": reference, "reason": "terminal queue evidence is incomplete"}
203 )
204 for record in ctx.checkpoint.state.get("jobs", []):
205 if not record.get("deleted"):
206 name = str(record.get("k8s_job_name") or record["name"])
207 namespace = str(record.get("k8s_job_namespace") or record["namespace"])
208 reference = f"{record['region']}:{namespace}/{name}"
209 if not any(item["resource"] == reference for item in result["unresolved"]):
210 result["unresolved"].append(
211 {"resource": reference, "reason": "UID-bound Job absence is incomplete"}
212 )
214 result["complete"] = not result["errors"] and not result["unresolved"]
215 result["ended_at"] = utc_now()
216 ctx.checkpoint.state.setdefault("workload_cleanup_attempts", []).append(result)
217 ctx.persist()
218 return result