Coverage for scripts / live_release_validation / checks / inference_runtime.py: 100.00%

1"""Readiness and HPA stability checks for live inference endpoints.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import subprocess 

7import time 

8from collections.abc import Callable 

9from typing import Any, cast 

10 

11from .inference_common import ManagedInferenceValidationError 

12 

13_TUNNEL_HEARTBEAT_INTERVAL_SECONDS = 240.0 

14 

15# Workload diagnostics: what the endpoint's pods are doing while a phase waits 

16# on DDB or Kubernetes. A crash-looping SGLang pod once ran out the whole 

17# 30-minute readiness window with the checkpoint recording nothing but the 

18# timeout; the cause (eight restarts, "no kernel image is available for 

19# execution on the device" on a T4 node) had to be dug out of CloudWatch by 

20# hand after the cluster was gone. Snapshots are taken every 

21# _DIAGNOSTICS_INTERVAL_SECONDS during a wait and once more when it fails, 

22# kept as a bounded ring so a long wait cannot bloat the checkpoint. 

23_DIAGNOSTICS_INTERVAL_SECONDS = 300.0 

24_DIAGNOSTICS_RING_SIZE = 8 

25_DIAGNOSTICS_LOG_TAIL_LINES = 40 

26_DIAGNOSTICS_LOG_TAIL_BYTES = 6 * 1024 

27_DIAGNOSTICS_EVENT_LIMIT = 20 

28_DIAGNOSTICS_NODE_LABELS = ( 

29 "node.kubernetes.io/instance-type", 

30 "eks.amazonaws.com/instance-family", 

31 "eks.amazonaws.com/instance-gpu-name", 

32 "eks.amazonaws.com/instance-gpu-count", 

33 "eks.amazonaws.com/instance-gpu-memory", 

34 "karpenter.sh/nodepool", 

35 "eks.amazonaws.com/nodepool", 

36 "karpenter.sh/capacity-type", 

37 "eks.amazonaws.com/capacity-type", 

38 "topology.kubernetes.io/zone", 

39) 

40 

41 

42def _dict_or_empty(value: Any) -> dict[str, Any]: 

43 """A dict payload as itself, anything else as an empty dict.""" 

44 return value if isinstance(value, dict) else {} 

45 

46 

47class InferenceRuntimeMixin: 

48 """Mixin for DDB readiness, Kubernetes readiness, and HPA stability.""" 

49 

50 settings: Any 

51 

52 def _persist(self) -> None: # pragma: no cover - implemented by lifecycle 

53 raise NotImplementedError 

54 

55 def _strong_get( 

56 self, record: dict[str, Any] 

57 ) -> dict[str, Any] | None: # pragma: no cover - implemented by lifecycle 

58 raise NotImplementedError 

59 

60 def _is_owned( 

61 self, item: dict[str, Any] 

62 ) -> bool: # pragma: no cover - implemented by lifecycle 

63 raise NotImplementedError 

64 

65 def _verify_item_contract( 

66 self, plan: Any, item: dict[str, Any], record: dict[str, Any] 

67 ) -> None: # pragma: no cover - implemented by lifecycle 

68 raise NotImplementedError 

69 

70 def _set_phase( 

71 self, record: dict[str, Any], phase: str, **values: Any 

72 ) -> None: # pragma: no cover - implemented by lifecycle 

73 raise NotImplementedError 

74 

75 _kubectl_json: Callable[..., Any | None] 

76 kubectl: Callable[..., tuple[int, str, str]] 

77 

78 def keep_cluster_tunnel_alive( 

79 self, 

80 record: dict[str, Any], 

81 last_heartbeat: float, 

82 *, 

83 deadline: float | None = None, 

84 ) -> float: 

85 """Send bounded Kubernetes traffic before SSM's idle-session timeout.""" 

86 now = time.monotonic() 

87 if now - last_heartbeat < _TUNNEL_HEARTBEAT_INTERVAL_SECONDS: 

88 return last_heartbeat 

89 process_timeout = 8.0 

90 if deadline is not None: 

91 remaining = deadline - now 

92 if remaining <= 0: 

93 raise ManagedInferenceValidationError( 

94 "managed inference tunnel heartbeat deadline expired" 

95 ) 

96 process_timeout = min(process_timeout, remaining) 

97 observation: dict[str, Any] = {"started_at_monotonic": now} 

98 try: 

99 returncode, stdout, stderr = self.kubectl( 

100 "--request-timeout=5s", 

101 "get", 

102 "--raw=/readyz", 

103 timeout=process_timeout, 

104 ) 

105 except (OSError, subprocess.TimeoutExpired) as exc: 

106 observation.update( 

107 { 

108 "healthy": False, 

109 "error": f"{type(exc).__name__}: {exc}", 

110 } 

111 ) 

112 else: 

113 observation.update( 

114 { 

115 "healthy": returncode == 0 and stdout.strip() == "ok", 

116 "returncode": returncode, 

117 "stderr": stderr[-1000:], 

118 } 

119 ) 

120 history = record.setdefault("tunnel_heartbeats", []) 

121 if not isinstance(history, list): 

122 raise ManagedInferenceValidationError("managed tunnel heartbeat history is invalid") 

123 history.append(observation) 

124 self._persist() 

125 if observation["healthy"] is not True: 

126 raise ManagedInferenceValidationError( 

127 "managed inference Kubernetes tunnel heartbeat failed" 

128 ) 

129 return time.monotonic() 

130 

131 def _wait_for_owned_record( 

132 self, 

133 plan: Any, 

134 record: dict[str, Any], 

135 ) -> dict[str, Any]: 

136 """Wait for the run-owned DDB record while keeping the tunnel active.""" 

137 deadline = time.monotonic() + self.settings.readiness_timeout_seconds 

138 heartbeat_at = float("-inf") 

139 while True: 

140 if time.monotonic() >= deadline: 

141 raise ManagedInferenceValidationError( 

142 "managed inference endpoint ownership did not appear before timeout" 

143 ) 

144 item = self._strong_get(record) 

145 if item is not None: 

146 if not self._is_owned(item): 

147 raise ManagedInferenceValidationError( 

148 "managed inference endpoint collision detected; refusing ownership" 

149 ) 

150 self._verify_item_contract(plan, item, record) 

151 record["owned"] = True 

152 self._set_phase(record, "ownership-confirmed") 

153 return item 

154 heartbeat_at = self.keep_cluster_tunnel_alive( 

155 record, 

156 heartbeat_at, 

157 deadline=deadline, 

158 ) 

159 remaining = deadline - time.monotonic() 

160 if remaining <= 0: 

161 raise ManagedInferenceValidationError( 

162 "managed inference endpoint ownership did not appear before timeout" 

163 ) 

164 time.sleep(min(float(self.settings.poll_interval_seconds), remaining)) 

165 

166 # ------------------------------------------------------------------ 

167 # Workload diagnostics 

168 # ------------------------------------------------------------------ 

169 

170 def _diagnostics_read(self, *arguments: str) -> tuple[Any | None, str | None]: 

171 """One best-effort kubectl read for a diagnostics snapshot. 

172 

173 Returns ``(payload, error)``; JSON output is decoded, anything else is 

174 returned as text. Never raises: a snapshot that cannot be taken is 

175 itself evidence and must not mask the failure being diagnosed. 

176 """ 

177 timeout = min(float(self.settings.command_timeout_seconds), 30.0) 

178 try: 

179 code, stdout, stderr = self.kubectl(*arguments, timeout=timeout) 

180 except Exception as exc: 

181 return None, f"{type(exc).__name__}: {exc}" 

182 if code != 0: 

183 return None, f"kubectl exited {code}: {stderr.strip()[-500:]}" 

184 if "json" in arguments: 

185 try: 

186 return json.loads(stdout), None 

187 except json.JSONDecodeError as exc: 

188 return None, f"kubectl returned non-JSON output: {exc}" 

189 return stdout, None 

190 

191 @staticmethod 

192 def _summarize_container_state(state: Any) -> dict[str, Any] | None: 

193 """Flatten a container ``state``/``lastState`` block to its one active branch.""" 

194 if not isinstance(state, dict): 

195 return None 

196 for branch in ("waiting", "running", "terminated"): 

197 details = state.get(branch) 

198 if isinstance(details, dict): 

199 summary: dict[str, Any] = {"status": branch} 

200 for key in ("reason", "exitCode", "signal", "startedAt", "finishedAt"): 

201 if details.get(key) is not None: 

202 summary[key] = details[key] 

203 message = details.get("message") 

204 if isinstance(message, str) and message: 

205 summary["message"] = message[-500:] 

206 return summary 

207 return None 

208 

209 def _diagnose_pod(self, plan: Any, pod: dict[str, Any]) -> dict[str, Any]: 

210 """Describe one pod: scheduling, per-container state, and crash output.""" 

211 metadata = _dict_or_empty(pod.get("metadata")) 

212 spec = _dict_or_empty(pod.get("spec")) 

213 status = _dict_or_empty(pod.get("status")) 

214 name = str(metadata.get("name") or "") 

215 diagnosis: dict[str, Any] = { 

216 "name": name, 

217 "phase": status.get("phase"), 

218 "node": spec.get("nodeName"), 

219 "start_time": status.get("startTime"), 

220 "conditions": [], 

221 "containers": [], 

222 } 

223 conditions = status.get("conditions") 

224 if isinstance(conditions, list): 

225 for condition in conditions: 

226 if not isinstance(condition, dict) or condition.get("status") == "True": 

227 continue 

228 condition_entry: dict[str, Any] = { 

229 "type": condition.get("type"), 

230 "status": condition.get("status"), 

231 } 

232 if condition.get("reason"): 

233 condition_entry["reason"] = condition["reason"] 

234 if isinstance(condition.get("message"), str): 

235 condition_entry["message"] = condition["message"][-500:] 

236 diagnosis["conditions"].append(condition_entry) 

237 statuses = status.get("containerStatuses") 

238 if isinstance(statuses, list): 

239 for container_status in statuses: 

240 if not isinstance(container_status, dict): 

241 continue 

242 container_name = str(container_status.get("name") or "") 

243 restarts = int(container_status.get("restartCount") or 0) 

244 ready = container_status.get("ready") is True 

245 entry: dict[str, Any] = { 

246 "name": container_name, 

247 "ready": ready, 

248 "restart_count": restarts, 

249 "state": self._summarize_container_state(container_status.get("state")), 

250 "last_state": self._summarize_container_state( 

251 container_status.get("lastState") 

252 ), 

253 } 

254 if name and container_name and (restarts > 0 or not ready): 

255 # The current attempt's output, and — the part that names 

256 # a crash-loop's cause — the previous attempt's. 

257 entry["log_tail"] = self._log_tail(name, container_name, previous=False) 

258 if restarts > 0: 

259 entry["previous_log_tail"] = self._log_tail( 

260 name, container_name, previous=True 

261 ) 

262 diagnosis["containers"].append(entry) 

263 return diagnosis 

264 

265 def _log_tail(self, pod: str, container: str, *, previous: bool) -> str: 

266 arguments = [ 

267 "logs", 

268 pod, 

269 "--namespace", 

270 self.settings.namespace, 

271 "--container", 

272 container, 

273 f"--tail={_DIAGNOSTICS_LOG_TAIL_LINES}", 

274 ] 

275 if previous: 

276 arguments.append("--previous") 

277 payload, error = self._diagnostics_read(*arguments) 

278 if error is not None: 

279 return f"<unavailable: {error}>" 

280 return str(payload)[-_DIAGNOSTICS_LOG_TAIL_BYTES:] 

281 

282 def _diagnose_events(self, plan: Any) -> list[dict[str, Any]] | str: 

283 payload, error = self._diagnostics_read( 

284 "get", 

285 "events", 

286 "--namespace", 

287 self.settings.namespace, 

288 "--output", 

289 "json", 

290 ) 

291 if error is not None: 

292 return f"<unavailable: {error}>" 

293 items = payload.get("items") if isinstance(payload, dict) else None 

294 events: list[dict[str, Any]] = [] 

295 for item in items if isinstance(items, list) else []: 

296 if not isinstance(item, dict): 

297 continue 

298 involved = item.get("involvedObject") 

299 involved_name = str(involved.get("name") or "") if isinstance(involved, dict) else "" 

300 if not involved_name.startswith(str(plan.name)): 

301 continue 

302 message = item.get("message") 

303 events.append( 

304 { 

305 "object": ( 

306 f"{involved.get('kind')}/{involved_name}" 

307 if isinstance(involved, dict) 

308 else involved_name 

309 ), 

310 "type": item.get("type"), 

311 "reason": item.get("reason"), 

312 "count": item.get("count"), 

313 "last_timestamp": item.get("lastTimestamp") or item.get("eventTime"), 

314 "message": message[-500:] if isinstance(message, str) else message, 

315 } 

316 ) 

317 events.sort(key=lambda event: str(event.get("last_timestamp") or "")) 

318 return events[-_DIAGNOSTICS_EVENT_LIMIT:] 

319 

320 def _diagnose_node(self, node_name: str) -> dict[str, Any]: 

321 payload, error = self._diagnostics_read("get", "node", node_name, "--output", "json") 

322 if error is not None: 

323 return {"error": error} 

324 metadata = payload.get("metadata") if isinstance(payload, dict) else None 

325 labels = metadata.get("labels") if isinstance(metadata, dict) else None 

326 status = payload.get("status") if isinstance(payload, dict) else None 

327 allocatable = status.get("allocatable") if isinstance(status, dict) else None 

328 return { 

329 "labels": { 

330 key: labels[key] 

331 for key in _DIAGNOSTICS_NODE_LABELS 

332 if isinstance(labels, dict) and key in labels 

333 }, 

334 "allocatable_gpus": ( 

335 allocatable.get("nvidia.com/gpu") if isinstance(allocatable, dict) else None 

336 ), 

337 } 

338 

339 @staticmethod 

340 def _diagnostics_summary(snapshot: dict[str, Any]) -> str: 

341 """One line a human can act on: pods, their worst container state, their GPU.""" 

342 pods = snapshot.get("pods") 

343 if not isinstance(pods, list) or not pods: 

344 error = snapshot.get("pods_error") 

345 return f"no pods observed{f' ({error})' if error else ''}" 

346 parts: list[str] = [] 

347 nodes = _dict_or_empty(snapshot.get("nodes")) 

348 for pod in pods: 

349 phase = pod.get("phase") or "?" 

350 pieces = [f"{pod.get('name')}: {phase}"] 

351 for condition in pod.get("conditions") or []: 

352 if condition.get("type") == "PodScheduled" and condition.get("reason"): 

353 pieces.append(f"unscheduled ({condition['reason']})") 

354 for container in pod.get("containers") or []: 

355 state = container.get("state") or {} 

356 last = container.get("last_state") or {} 

357 detail = state.get("reason") or state.get("status") or "?" 

358 if last.get("exitCode") is not None: 

359 detail += f", last exit {last['exitCode']}" 

360 if last.get("reason"): 

361 detail += f" {last['reason']}" 

362 pieces.append( 

363 f"{container.get('name')} {detail} (restarts={container.get('restart_count', 0)})" 

364 ) 

365 node = nodes.get(pod.get("node") or "") 

366 if isinstance(node, dict) and isinstance(node.get("labels"), dict): 

367 labels = node["labels"] 

368 placement = "/".join( 

369 str(labels[key]) 

370 for key in ( 

371 "node.kubernetes.io/instance-type", 

372 "eks.amazonaws.com/instance-gpu-name", 

373 ) 

374 if key in labels 

375 ) 

376 if placement: 

377 pieces.append(f"on {placement}") 

378 parts.append(", ".join(pieces)) 

379 return "; ".join(parts) 

380 

381 def capture_workload_diagnostics( 

382 self, 

383 plan: Any, 

384 record: dict[str, Any], 

385 *, 

386 reason: str, 

387 ) -> dict[str, Any]: 

388 """Snapshot the endpoint's pods, crash output, events and nodes into the record. 

389 

390 Best-effort and bounded: every read has its own short timeout and a 

391 failed read is recorded in place of the data. The snapshot ring keeps 

392 the last ``_DIAGNOSTICS_RING_SIZE`` entries so periodic captures during 

393 a long wait show the progression (pulling -> running -> crash-looping) 

394 without growing the checkpoint unboundedly. Returns the snapshot; its 

395 ``summary`` is what the caller appends to a timeout error. 

396 """ 

397 snapshot: dict[str, Any] = { 

398 "reason": reason, 

399 "captured_at_monotonic": time.monotonic(), 

400 "pods": [], 

401 "nodes": {}, 

402 } 

403 payload, error = self._diagnostics_read( 

404 "get", 

405 "pods", 

406 "--namespace", 

407 self.settings.namespace, 

408 "--selector", 

409 f"app={plan.name}", 

410 "--output", 

411 "json", 

412 ) 

413 if error is not None: 

414 snapshot["pods_error"] = error 

415 else: 

416 items = payload.get("items") if isinstance(payload, dict) else None 

417 for pod in items if isinstance(items, list) else []: 

418 if isinstance(pod, dict): 

419 snapshot["pods"].append(self._diagnose_pod(plan, pod)) 

420 for node_name in sorted({str(pod["node"]) for pod in snapshot["pods"] if pod["node"]}): 

421 snapshot["nodes"][node_name] = self._diagnose_node(node_name) 

422 snapshot["events"] = self._diagnose_events(plan) 

423 snapshot["summary"] = self._diagnostics_summary(snapshot) 

424 

425 ring = record.setdefault("workload_diagnostics", []) 

426 if not isinstance(ring, list): 

427 ring = [] 

428 record["workload_diagnostics"] = ring 

429 ring.append(snapshot) 

430 del ring[:-_DIAGNOSTICS_RING_SIZE] 

431 record["last_workload_summary"] = snapshot["summary"] 

432 self._persist() 

433 return snapshot 

434 

435 def _diagnostics_due( 

436 self, record: dict[str, Any], plan: Any, last_capture: float, reason: str 

437 ) -> float: 

438 """Take a periodic snapshot when the interval has elapsed; return the new mark.""" 

439 now = time.monotonic() 

440 if now - last_capture < _DIAGNOSTICS_INTERVAL_SECONDS: 

441 return last_capture 

442 self.capture_workload_diagnostics(plan, record, reason=reason) 

443 return now 

444 

445 def _timeout_error( 

446 self, plan: Any, record: dict[str, Any], message: str, *, reason: str 

447 ) -> ManagedInferenceValidationError: 

448 """Build a timeout error that carries the final workload diagnosis.""" 

449 snapshot = self.capture_workload_diagnostics(plan, record, reason=reason) 

450 return ManagedInferenceValidationError(f"{message} ({snapshot['summary']})") 

451 

452 def wait_for_ddb_running(self, plan: Any, record: dict[str, Any]) -> None: 

453 """Require this run's exact DDB record and running regional observation.""" 

454 deadline = time.monotonic() + self.settings.readiness_timeout_seconds 

455 heartbeat_at = float("-inf") 

456 diagnostics_at = time.monotonic() 

457 while True: 

458 if time.monotonic() >= deadline: 

459 raise self._timeout_error( 

460 plan, 

461 record, 

462 "managed inference DDB running state was not observed before timeout", 

463 reason="ddb-running-timeout", 

464 ) 

465 item = self._strong_get(record) 

466 if item is not None: 

467 if not self._is_owned(item): 

468 raise ManagedInferenceValidationError( 

469 "managed inference endpoint ownership changed while waiting" 

470 ) 

471 self._verify_item_contract(plan, item, record) 

472 statuses = item.get("region_status") 

473 regional = ( 

474 statuses.get(self.settings.selected_region) 

475 if isinstance(statuses, dict) 

476 else None 

477 ) 

478 # The monitor's own view of the region — its state and any 

479 # message it wrote — is the cheapest evidence there is. 

480 record["last_ddb_observation"] = { 

481 "desired_state": item.get("desired_state"), 

482 "regional": regional if isinstance(regional, dict) else None, 

483 "observed_at_monotonic": time.monotonic(), 

484 } 

485 if ( 

486 item.get("desired_state") == "running" 

487 and isinstance(regional, dict) 

488 and regional.get("state") == "running" 

489 and self._regional_status_satisfied(plan, record, regional) 

490 ): 

491 self._set_phase(record, "ddb-running") 

492 return 

493 heartbeat_at = self.keep_cluster_tunnel_alive( 

494 record, 

495 heartbeat_at, 

496 deadline=deadline, 

497 ) 

498 diagnostics_at = self._diagnostics_due(record, plan, diagnostics_at, "ddb-running-wait") 

499 remaining = deadline - time.monotonic() 

500 if remaining <= 0: 

501 raise self._timeout_error( 

502 plan, 

503 record, 

504 "managed inference DDB running state was not observed before timeout", 

505 reason="ddb-running-timeout", 

506 ) 

507 time.sleep(min(float(self.settings.poll_interval_seconds), remaining)) 

508 

509 def _regional_status_satisfied( 

510 self, plan: Any, record: dict[str, Any], regional: dict[str, Any] 

511 ) -> bool: 

512 """Scenario-specific conditions on the monitor's regional observation. 

513 

514 A Mooncake store scenario is ready only when the shared master is 

515 reported Ready and the requested cold tier is active with the 

516 own-Region bucket URI: an unresolved cold tier (the monitor dropped it 

517 and reported why) is not a running store for this harness, since the 

518 scenario exists to prove that tier end to end. 

519 """ 

520 if not getattr(plan, "mooncake_store", False): 

521 return True 

522 store = _dict_or_empty(regional.get("store")) 

523 cold_tier = _dict_or_empty(store.get("cold_tier")) 

524 s3_uri = cold_tier.get("s3_uri") 

525 satisfied = ( 

526 store.get("ready") is True 

527 and cold_tier.get("enabled") is True 

528 and isinstance(s3_uri, str) 

529 and s3_uri.startswith("s3://") 

530 ) 

531 record["last_store_observation"] = { 

532 "ready": store.get("ready"), 

533 "master": store.get("master"), 

534 "cold_tier": cold_tier, 

535 "satisfied": satisfied, 

536 } 

537 if satisfied: 

538 record["cold_tier_verified"] = True 

539 return satisfied 

540 

541 @staticmethod 

542 def _ready_condition(container_statuses: Any) -> bool: 

543 return ( 

544 isinstance(container_statuses, list) 

545 and bool(container_statuses) 

546 and all( 

547 isinstance(status, dict) and status.get("ready") is True 

548 for status in container_statuses 

549 ) 

550 ) 

551 

552 def _deployment_ready_snapshot( 

553 self, 

554 plan: Any, 

555 record: dict[str, Any], 

556 expected_replicas: int, 

557 *, 

558 exact: bool, 

559 deadline: float | None = None, 

560 ) -> tuple[bool, dict[str, int]]: 

561 deployment = self._kubectl_json( 

562 record, 

563 "get", 

564 "deployment", 

565 plan.name, 

566 "--namespace", 

567 self.settings.namespace, 

568 "--output", 

569 "json", 

570 deadline=deadline, 

571 ) 

572 if not isinstance(deployment, dict): 

573 return False, {} 

574 metadata = deployment.get("metadata") 

575 spec = deployment.get("spec") 

576 status = deployment.get("status") 

577 if ( 

578 not isinstance(metadata, dict) 

579 or not isinstance(spec, dict) 

580 or not isinstance(status, dict) 

581 ): 

582 return False, {} 

583 desired = int(spec.get("replicas") or 0) 

584 ready = int(status.get("readyReplicas") or 0) 

585 available = int(status.get("availableReplicas") or 0) 

586 updated = int(status.get("updatedReplicas") or 0) 

587 generation = int(metadata.get("generation") or 0) 

588 observed = int(status.get("observedGeneration") or 0) 

589 replica_match = desired == expected_replicas if exact else desired >= expected_replicas 

590 deployment_ready = ( 

591 replica_match 

592 and ready >= expected_replicas 

593 and available >= expected_replicas 

594 and updated >= expected_replicas 

595 and observed >= generation 

596 ) 

597 

598 pods_payload = self._kubectl_json( 

599 record, 

600 "get", 

601 "pods", 

602 "--namespace", 

603 self.settings.namespace, 

604 "--selector", 

605 f"app={plan.name}", 

606 "--output", 

607 "json", 

608 deadline=deadline, 

609 ) 

610 items = pods_payload.get("items", []) if isinstance(pods_payload, dict) else [] 

611 ready_pods = 0 

612 if isinstance(items, list): 

613 for item in items: 

614 pod_status = item.get("status") if isinstance(item, dict) else None 

615 if ( 

616 isinstance(pod_status, dict) 

617 and pod_status.get("phase") == "Running" 

618 and self._ready_condition(pod_status.get("containerStatuses")) 

619 ): 

620 ready_pods += 1 

621 evidence = { 

622 "desired": desired, 

623 "ready": ready, 

624 "available": available, 

625 "updated": updated, 

626 "ready_pods": ready_pods, 

627 } 

628 return deployment_ready and ready_pods >= expected_replicas, evidence 

629 

630 def wait_for_kubernetes_ready(self, plan: Any, record: dict[str, Any]) -> None: 

631 """Require Deployment convergence and ready Running pods.""" 

632 deadline = time.monotonic() + self.settings.readiness_timeout_seconds 

633 diagnostics_at = time.monotonic() 

634 while True: 

635 ready, evidence = self._deployment_ready_snapshot( 

636 plan, 

637 record, 

638 plan.replicas, 

639 exact=not plan.autoscaling, 

640 deadline=deadline, 

641 ) 

642 record["last_readiness"] = evidence 

643 self._persist() 

644 if ready: 

645 self._set_phase(record, "kubernetes-ready") 

646 return 

647 if time.monotonic() >= deadline: 

648 raise self._timeout_error( 

649 plan, 

650 record, 

651 "managed inference Kubernetes readiness was not observed before timeout", 

652 reason="kubernetes-ready-timeout", 

653 ) 

654 diagnostics_at = self._diagnostics_due( 

655 record, plan, diagnostics_at, "kubernetes-ready-wait" 

656 ) 

657 time.sleep( 

658 min( 

659 float(self.settings.poll_interval_seconds), 

660 max(0.0, deadline - time.monotonic()), 

661 ) 

662 ) 

663 

664 def verify_no_container_restarts(self, plan: Any, record: dict[str, Any]) -> None: 

665 """Refuse a leg whose containers restarted on the way to serving. 

666 

667 Readiness, HPA stability and a successful invocation can all be 

668 observed *between* restarts: an SGLang container whose probes timed 

669 out was killed by liveness every ~3 minutes while readiness flapped 

670 true often enough for the leg to pass (2026-09-18, A10G). A restart 

671 means a crash or a probe kill, and either is a defect the release 

672 must not carry, so the final diagnostics snapshot is audited: every 

673 container of every pod must report ``restartCount`` 0. The snapshot 

674 already holds the previous attempt's log tail and the pod events 

675 (``Liveness probe failed``, ``Killing``), so the failure names its 

676 cause. A snapshot that could not list the pods fails closed. 

677 """ 

678 snapshot = self.capture_workload_diagnostics(plan, record, reason="restart-audit") 

679 if "pods_error" in snapshot: 

680 raise ManagedInferenceValidationError( 

681 "managed inference restart audit could not list the endpoint's pods " 

682 f"({snapshot['pods_error']})" 

683 ) 

684 pods = snapshot["pods"] 

685 if not pods: 

686 raise ManagedInferenceValidationError( 

687 "managed inference restart audit found no pods for a serving endpoint" 

688 ) 

689 restarted = [ 

690 f"{pod.get('name')}/{container.get('name')} restarted " 

691 f"{int(container.get('restart_count') or 0)}x" 

692 for pod in pods 

693 for container in pod.get("containers") or [] 

694 if int(container.get("restart_count") or 0) > 0 

695 ] 

696 record["restart_audit"] = { 

697 "pods": [str(pod.get("name")) for pod in pods], 

698 "restarted": restarted, 

699 } 

700 self._persist() 

701 if restarted: 

702 raise ManagedInferenceValidationError( 

703 "managed inference containers restarted during the leg: " 

704 f"{'; '.join(restarted)} ({snapshot['summary']})" 

705 ) 

706 self._set_phase(record, "restart-audited") 

707 

708 def _hpa_matches( 

709 self, 

710 plan: Any, 

711 record: dict[str, Any], 

712 *, 

713 deadline: float | None = None, 

714 ) -> bool: 

715 hpa = self._kubectl_json( 

716 record, 

717 "get", 

718 "horizontalpodautoscaler.autoscaling", 

719 plan.name, 

720 "--namespace", 

721 self.settings.namespace, 

722 "--output", 

723 "json", 

724 deadline=deadline, 

725 ) 

726 if not isinstance(hpa, dict): 

727 return False 

728 spec = hpa.get("spec") 

729 if not isinstance(spec, dict): 

730 return False 

731 target = spec.get("scaleTargetRef") 

732 if not isinstance(target, dict) or target != { 

733 "apiVersion": "apps/v1", 

734 "kind": "Deployment", 

735 "name": plan.name, 

736 }: 

737 return False 

738 if spec.get("minReplicas") != self.settings.hpa_min_replicas: 

739 return False 

740 if spec.get("maxReplicas") != self.settings.hpa_max_replicas: 

741 return False 

742 metrics = spec.get("metrics") 

743 if not isinstance(metrics, list): 

744 return False 

745 return any( 

746 isinstance(metric, dict)