Coverage for gco / services / api_routes / jobs.py: 100.00%

1"""Job listing, details, logs, events, metrics, delete, and retry endpoints.""" 

2 

3from __future__ import annotations 

4 

5import copy 

6import json 

7import logging 

8import re 

9from datetime import UTC, datetime, timedelta 

10from typing import Any 

11 

12from fastapi import APIRouter, HTTPException, Query 

13from fastapi.responses import JSONResponse, Response 

14from kubernetes import client as kubernetes_client 

15 

16from gco.k8s_api_versions import METRICS_API_GROUP, METRICS_API_VERSION 

17from gco.models import ManifestSubmissionRequest 

18from gco.services.api_shared import ( 

19 BulkDeleteRequest, 

20 _check_namespace, 

21 _check_processor, 

22 _collect_pod_scheduling, 

23 _empty_scheduling_info, 

24 _parse_event_to_dict, 

25 _parse_job_to_dict, 

26 _parse_pod_to_dict, 

27 internal_server_error, 

28) 

29from gco.services.structured_logging import sanitize_log_value 

30 

31router = APIRouter(prefix="/api/v1/jobs", tags=["Jobs"]) 

32logger = logging.getLogger(__name__) 

33 

34_LABEL_NAME_RE = re.compile(r"^[A-Za-z0-9](?:[-_.A-Za-z0-9]{0,61}[A-Za-z0-9])?$") 

35_DNS_LABEL_RE = re.compile(r"^[a-z0-9](?:[-a-z0-9]{0,61}[a-z0-9])?$") 

36 

37 

38def _decode_pod_log_response(response: Any) -> str: 

39 """Decode the raw Kubernetes log body without converting bytes to their repr.""" 

40 release_conn = getattr(response, "release_conn", None) 

41 try: 

42 payload = response.data if hasattr(response, "data") else response 

43 if isinstance(payload, str): 

44 return payload 

45 if isinstance(payload, (bytes, bytearray)): 

46 return bytes(payload).decode("utf-8", errors="replace") 

47 raise TypeError(f"Kubernetes returned an unsupported pod log payload: {type(payload)!r}") 

48 finally: 

49 if callable(release_conn): 

50 release_conn() 

51 

52 

53def _parse_exact_label_selector(selector: str | None) -> list[tuple[str, str]]: 

54 """Parse the API's deliberately narrow, fail-closed selector subset.""" 

55 if selector is None: 

56 return [] 

57 

58 requirements: list[tuple[str, str]] = [] 

59 for raw_clause in selector.split(","): 

60 clause = raw_clause.strip() 

61 if not clause or clause.count("=") != 1: 

62 raise ValueError( 

63 "Label selectors must be comma-separated exact matches in key=value form" 

64 ) 

65 

66 key, value = (part.strip() for part in clause.split("=", 1)) 

67 if "/" in key: 

68 if key.count("/") != 1: 

69 raise ValueError(f"Invalid label key in selector: {key!r}") 

70 prefix, name = key.split("/", 1) 

71 valid_prefix = len(prefix) <= 253 and all( 

72 _DNS_LABEL_RE.fullmatch(part) for part in prefix.split(".") 

73 ) 

74 else: 

75 name = key 

76 valid_prefix = True 

77 

78 if not valid_prefix or not _LABEL_NAME_RE.fullmatch(name): 

79 raise ValueError(f"Invalid label key in selector: {key!r}") 

80 if value and not _LABEL_NAME_RE.fullmatch(value): 

81 raise ValueError(f"Invalid label value in selector for {key!r}") 

82 requirements.append((key, value)) 

83 

84 return requirements 

85 

86 

87def _labels_match(labels: Any, requirements: list[tuple[str, str]]) -> bool: 

88 """Return whether all parsed exact-match requirements are satisfied.""" 

89 if not isinstance(labels, dict): 

90 return False 

91 return all(labels.get(key) == value for key, value in requirements) 

92 

93 

94# Labels the API server stamps on a Job's pod template (and on the Job itself 

95# when it was created without labels of its own). They carry the original 

96# Job's UID and name, so a copy that keeps them fails the generated-selector 

97# validation of the new Job. 

98_JOB_CONTROLLER_LABELS = frozenset( 

99 { 

100 "controller-uid", 

101 "job-name", 

102 "batch.kubernetes.io/controller-uid", 

103 "batch.kubernetes.io/job-name", 

104 } 

105) 

106# Annotations that describe the original object rather than the workload. 

107_NON_PORTABLE_JOB_ANNOTATIONS = frozenset({"kubectl.kubernetes.io/last-applied-configuration"}) 

108_RETRY_NAME_SUFFIX_FORMAT = "-retry-%Y%m%d%H%M%S" 

109# The Job controller copies the Job name into its pods' job-name label, so the 

110# API server rejects Job names longer than a label value. 

111_MAX_JOB_NAME_LENGTH = 63 

112 

113 

114def _retry_job_name(name: str, now: datetime) -> str: 

115 """Name a retry, shortening the original name so the result fits a Job.""" 

116 suffix = now.strftime(_RETRY_NAME_SUFFIX_FORMAT) 

117 prefix = name[: _MAX_JOB_NAME_LENGTH - len(suffix)].rstrip("-.") 

118 return f"{prefix}{suffix}" 

119 

120 

121def _read_job_object(processor: Any, name: str, namespace: str) -> dict[str, Any]: 

122 """Read a Job exactly as the API server returns it. 

123 

124 The typed ``V1Job`` model is the wrong source for a copy: its 

125 ``to_dict()`` spells fields the Python way (``restart_policy``), which the 

126 API server does not recognise, and it drops any field newer than the 

127 client's models. 

128 """ 

129 response = processor.batch_v1.read_namespaced_job( 

130 name=name, namespace=namespace, _preload_content=False 

131 ) 

132 release_conn = getattr(response, "release_conn", None) 

133 try: 

134 job = json.loads(response.data) 

135 finally: 

136 if callable(release_conn): 

137 release_conn() 

138 if not isinstance(job, dict): 

139 raise TypeError(f"Kubernetes returned an unsupported Job payload: {type(job)!r}") 

140 return job 

141 

142 

143def _retry_job_manifest( 

144 original: dict[str, Any], *, original_name: str, new_name: str, namespace: str 

145) -> dict[str, Any]: 

146 """Build a new Job from the original Job object. 

147 

148 The user's spec, labels and annotations carry over unchanged. What the API 

149 server generated for the original does not: its identity and status, the 

150 suspend flag, and, unless the Job manages its own selector 

151 (``manualSelector: true``), the generated selector and the 

152 controller-uid/job-name labels that tie pods to the original Job. 

153 """ 

154 metadata = original.get("metadata") or {} 

155 spec = copy.deepcopy(original.get("spec") or {}) 

156 generated_selector = spec.get("manualSelector") is not True 

157 

158 spec.pop("suspend", None) 

159 if generated_selector: 

160 spec.pop("selector", None) 

161 template = spec.get("template") 

162 template_metadata = template.get("metadata") if isinstance(template, dict) else None 

163 if isinstance(template_metadata, dict): 

164 template_metadata.pop("creationTimestamp", None) 

165 template_labels = template_metadata.get("labels") 

166 if generated_selector and isinstance(template_labels, dict): 

167 template_metadata["labels"] = { 

168 key: value 

169 for key, value in template_labels.items() 

170 if key not in _JOB_CONTROLLER_LABELS 

171 } 

172 

173 labels = { 

174 key: value 

175 for key, value in (metadata.get("labels") or {}).items() 

176 if not (generated_selector and key in _JOB_CONTROLLER_LABELS) 

177 } 

178 labels["gco.io/retry-of"] = original_name 

179 annotations = { 

180 key: value 

181 for key, value in (metadata.get("annotations") or {}).items() 

182 if key not in _NON_PORTABLE_JOB_ANNOTATIONS 

183 } 

184 annotations["gco.io/original-job"] = original_name 

185 

186 return { 

187 "apiVersion": "batch/v1", 

188 "kind": "Job", 

189 "metadata": { 

190 "name": new_name, 

191 "namespace": namespace, 

192 "labels": labels, 

193 "annotations": annotations, 

194 }, 

195 "spec": spec, 

196 } 

197 

198 

199@router.get("") 

200async def list_jobs( 

201 namespace: str | None = Query(None, description="Filter by namespace"), 

202 status: str | None = Query(None, description="Filter by status"), 

203 limit: int = Query(50, ge=1, le=1000, description="Maximum number of jobs to return"), 

204 offset: int = Query(0, ge=0, description="Number of jobs to skip"), 

205 sort: str = Query("createdAt:desc", description="Sort field and order (field:asc|desc)"), 

206 label_selector: str | None = Query( 

207 None, 

208 max_length=1024, 

209 description="Comma-separated exact-match label filters (key=value only)", 

210 ), 

211) -> Response: 

212 """List Kubernetes Jobs with pagination and filtering.""" 

213 processor = _check_processor() 

214 

215 try: 

216 selector_requirements = _parse_exact_label_selector(label_selector) 

217 all_jobs = await processor.list_jobs(namespace=namespace, status_filter=status) 

218 

219 if selector_requirements: 

220 all_jobs = [ 

221 job 

222 for job in all_jobs 

223 if _labels_match(job.get("metadata", {}).get("labels", {}), selector_requirements) 

224 ] 

225 

226 sort_field, sort_order = "createdAt", "desc" 

227 if ":" in sort: 

228 sort_field, sort_order = sort.split(":", 1) 

229 

230 def get_sort_key(job: dict[str, Any]) -> Any: 

231 if sort_field == "createdAt": 

232 return job.get("metadata", {}).get("creationTimestamp", "") 

233 if sort_field == "name": 

234 return job.get("metadata", {}).get("name", "") 

235 if sort_field == "status": 

236 return job.get("status", {}).get("active", 0) 

237 return "" 

238 

239 all_jobs.sort(key=get_sort_key, reverse=(sort_order == "desc")) 

240 

241 total = len(all_jobs) 

242 paginated_jobs = all_jobs[offset : offset + limit] 

243 

244 response = { 

245 "cluster_id": processor.cluster_id, 

246 "region": processor.region, 

247 "timestamp": datetime.now(UTC).isoformat(), 

248 "total": total, 

249 "limit": limit, 

250 "offset": offset, 

251 "has_more": (offset + limit) < total, 

252 "count": len(paginated_jobs), 

253 "jobs": paginated_jobs, 

254 } 

255 

256 return JSONResponse(status_code=200, content=response) 

257 

258 except ValueError as e: 

259 raise HTTPException(status_code=400, detail=str(e)) from e 

260 except Exception as e: 

261 raise internal_server_error("listing jobs", e) from e 

262 

263 

264def _job_scheduling(processor: Any, namespace: str, name: str) -> dict[str, Any]: 

265 """Report the nodes a Job's pods landed on, and each node's hardware. 

266 

267 Placement is supplementary to the Job read: a failure here (pods already 

268 garbage-collected, Node read refused) must never turn a successful job 

269 lookup into an error, so the failure is recorded in the payload instead. 

270 """ 

271 try: 

272 pods = processor.core_v1.list_namespaced_pod( 

273 namespace=namespace, label_selector=f"job-name={name}" 

274 ) 

275 return _collect_pod_scheduling(processor.core_v1, pods.items) 

276 except Exception as exc: 

277 # namespace/name arrive from the request path and the Kubernetes error 

278 # echoes them back, so all three are sanitized before logging (CWE-117). 

279 logger.warning( 

280 "Could not resolve node placement for %s/%s: %s", 

281 sanitize_log_value(namespace), 

282 sanitize_log_value(name), 

283 sanitize_log_value(exc), 

284 ) 

285 info = _empty_scheduling_info() 

286 info["node_lookup_error"] = str(exc) 

287 return info 

288 

289 

290@router.get("/{namespace}/{name}") 

291async def get_job(namespace: str, name: str) -> Response: 

292 """Get details of a specific Job, including where its pods were scheduled.""" 

293 processor = _check_processor() 

294 _check_namespace(namespace, processor) 

295 

296 try: 

297 job = processor.batch_v1.read_namespaced_job(name=name, namespace=namespace) 

298 job_info = _parse_job_to_dict(job) 

299 

300 response = { 

301 "cluster_id": processor.cluster_id, 

302 "region": processor.region, 

303 "timestamp": datetime.now(UTC).isoformat(), 

304 **job_info, 

305 "scheduling": _job_scheduling(processor, namespace, name), 

306 } 

307 

308 return JSONResponse(status_code=200, content=response) 

309 

310 except Exception as e: 

311 if "NotFound" in str(e) or "404" in str(e): 

312 raise HTTPException( 

313 status_code=404, detail=f"Job '{name}' not found in namespace '{namespace}'" 

314 ) from e 

315 raise internal_server_error("getting job", e) from e 

316 

317 

318@router.get("/{namespace}/{name}/logs") 

319async def get_job_logs( 

320 namespace: str, 

321 name: str, 

322 container: str | None = Query(None, description="Container name (for multi-container pods)"), 

323 tail: int = Query(100, ge=1, le=10000, description="Number of lines from the end"), 

324 previous: bool = Query(False, description="Get logs from previous terminated container"), 

325 since_seconds: int | None = Query( 

326 None, ge=1, description="Only return logs newer than N seconds" 

327 ), 

328 timestamps: bool = Query(False, description="Include timestamps in log lines"), 

329) -> Response: 

330 """Get logs from a Job's pods.""" 

331 from kubernetes.client.rest import ApiException as K8sApiException 

332 

333 processor = _check_processor() 

334 _check_namespace(namespace, processor) 

335 

336 try: 

337 try: 

338 processor.batch_v1.read_namespaced_job(name=name, namespace=namespace) 

339 except K8sApiException as e: 

340 if e.status == 404: 

341 raise HTTPException( 

342 status_code=404, 

343 detail=f"Job '{name}' not found in namespace '{namespace}'", 

344 ) from e 

345 raise 

346 

347 pods = processor.core_v1.list_namespaced_pod( 

348 namespace=namespace, label_selector=f"job-name={name}" 

349 ) 

350 

351 if not pods.items: 

352 raise HTTPException( 

353 status_code=404, 

354 detail=( 

355 f"No pods found for job '{name}'. " 

356 "The job may have completed and pods were cleaned up " 

357 "(ttlSecondsAfterFinished). Use 'gco jobs get' to check job status." 

358 ), 

359 ) 

360 

361 sorted_pods = sorted( 

362 pods.items, 

363 key=lambda p: p.metadata.creation_timestamp or datetime.min.replace(tzinfo=UTC), 

364 reverse=True, 

365 ) 

366 pod = sorted_pods[0] 

367 

368 pod_phase = pod.status.phase if pod.status else "Unknown" 

369 if pod_phase == "Pending": 

370 raise HTTPException( 

371 status_code=409, 

372 detail=( 

373 f"Pod '{pod.metadata.name}' is still Pending — logs are not yet available. " 

374 "The node may still be provisioning. Use 'gco jobs events' to check." 

375 ), 

376 ) 

377 

378 log_kwargs: dict[str, Any] = { 

379 "name": pod.metadata.name, 

380 "namespace": namespace, 

381 "tail_lines": tail, 

382 "previous": previous, 

383 "timestamps": timestamps, 

384 } 

385 if container: 

386 log_kwargs["container"] = container 

387 if since_seconds: 

388 log_kwargs["since_seconds"] = since_seconds 

389 

390 try: 

391 log_response = processor.core_v1.read_namespaced_pod_log( 

392 **log_kwargs, _preload_content=False 

393 ) 

394 logs = _decode_pod_log_response(log_response) 

395 except K8sApiException as e: 

396 if e.status == 400: 

397 error_body = str(e.body) if e.body else str(e.reason) 

398 if "waiting" in error_body.lower() or "not found" in error_body.lower(): 

399 available = [c.name for c in pod.spec.containers] 

400 raise HTTPException( 

401 status_code=400, 

402 detail=( 

403 f"Logs not available: {error_body}. " 

404 f"Pod phase: {pod_phase}. " 

405 f"Available containers: {available}" 

406 ), 

407 ) from e 

408 raise HTTPException(status_code=400, detail=f"Bad request: {error_body}") from e 

409 raise 

410 

411 available_containers = [c.name for c in pod.spec.containers] 

412 init_containers = [c.name for c in (pod.spec.init_containers or [])] 

413 

414 response = { 

415 "cluster_id": processor.cluster_id, 

416 "region": processor.region, 

417 "timestamp": datetime.now(UTC).isoformat(), 

418 "job_name": name, 

419 "namespace": namespace, 

420 "pod_name": pod.metadata.name, 

421 "container": container or (available_containers[0] if available_containers else None), 

422 "available_containers": available_containers, 

423 "init_containers": init_containers, 

424 "previous": previous, 

425 "tail_lines": tail, 

426 "logs": logs, 

427 } 

428 

429 return JSONResponse(status_code=200, content=response) 

430 

431 except HTTPException: 

432 raise 

433 except K8sApiException as e: 

434 logger.error(f"Kubernetes API error getting job logs: {e.status} {e.reason}") 

435 raise HTTPException( 

436 status_code=502, detail=f"Kubernetes API error: {e.status} {e.reason}" 

437 ) from e 

438 except Exception as e: 

439 raise internal_server_error("getting job logs", e) from e 

440 

441 

442@router.get("/{namespace}/{name}/events") 

443async def get_job_events(namespace: str, name: str) -> Response: 

444 """Get events related to a Job.""" 

445 processor = _check_processor() 

446 _check_namespace(namespace, processor) 

447 

448 try: 

449 field_selector = f"involvedObject.name={name},involvedObject.kind=Job" 

450 job_events = processor.core_v1.list_namespaced_event( 

451 namespace=namespace, field_selector=field_selector 

452 ) 

453 

454 pods = processor.core_v1.list_namespaced_pod( 

455 namespace=namespace, label_selector=f"job-name={name}" 

456 ) 

457 

458 pod_events = [] 

459 for pod in pods.items: 

460 field_selector = f"involvedObject.name={pod.metadata.name},involvedObject.kind=Pod" 

461 events = processor.core_v1.list_namespaced_event( 

462 namespace=namespace, field_selector=field_selector 

463 ) 

464 pod_events.extend(events.items) 

465 

466 all_events = [_parse_event_to_dict(e) for e in job_events.items] 

467 all_events.extend([_parse_event_to_dict(e) for e in pod_events]) 

468 all_events.sort( 

469 key=lambda e: e.get("lastTimestamp") or e.get("firstTimestamp") or "", reverse=True 

470 ) 

471 

472 response = { 

473 "cluster_id": processor.cluster_id, 

474 "region": processor.region, 

475 "timestamp": datetime.now(UTC).isoformat(), 

476 "job_name": name, 

477 "namespace": namespace, 

478 "count": len(all_events), 

479 "events": all_events, 

480 } 

481 

482 return JSONResponse(status_code=200, content=response) 

483 

484 except Exception as e: 

485 raise internal_server_error("getting job events", e) from e 

486 

487 

488@router.get("/{namespace}/{name}/pods") 

489async def get_job_pods(namespace: str, name: str) -> Response: 

490 """Get pods belonging to a Job, with the hardware each one landed on.""" 

491 processor = _check_processor() 

492 _check_namespace(namespace, processor) 

493 

494 try: 

495 pods = processor.core_v1.list_namespaced_pod( 

496 namespace=namespace, label_selector=f"job-name={name}" 

497 ) 

498 pod_list = [_parse_pod_to_dict(pod) for pod in pods.items] 

499 

500 try: 

501 scheduling = _collect_pod_scheduling(processor.core_v1, pods.items) 

502 except Exception as exc: # pragma: no cover - collector swallows its own 

503 logger.warning( 

504 "Could not resolve node placement for %s/%s: %s", 

505 sanitize_log_value(namespace), 

506 sanitize_log_value(name), 

507 sanitize_log_value(exc), 

508 ) 

509 scheduling = _empty_scheduling_info() 

510 scheduling["node_lookup_error"] = str(exc) 

511 

512 # Denormalized onto each pod so a caller iterating pods does not have to 

513 # join against the scheduling block to answer "what did this pod run on". 

514 node_index = {node["name"]: node for node in scheduling["nodes"]} 

515 for pod_entry in pod_list: 

516 node = node_index.get(pod_entry.get("spec", {}).get("nodeName")) 

517 pod_entry["node"] = ( 

518 { 

519 "name": node["name"], 

520 "instance_type": node["instance_type"], 

521 "capacity_type": node["capacity_type"], 

522 "labels": node["labels"], 

523 } 

524 if node 

525 else None 

526 ) 

527 

528 response = { 

529 "cluster_id": processor.cluster_id, 

530 "region": processor.region, 

531 "timestamp": datetime.now(UTC).isoformat(), 

532 "job_name": name, 

533 "namespace": namespace, 

534 "count": len(pod_list), 

535 "pods": pod_list, 

536 "scheduling": scheduling, 

537 } 

538 

539 return JSONResponse(status_code=200, content=response) 

540 

541 except Exception as e: 

542 raise internal_server_error("getting job pods", e) from e 

543 

544 

545@router.get("/{namespace}/{name}/pods/{pod_name}/logs") 

546async def get_pod_logs( 

547 namespace: str, 

548 name: str, 

549 pod_name: str, 

550 container: str | None = Query(None, description="Container name"), 

551 tail: int = Query(100, ge=1, le=10000, description="Number of lines from the end"), 

552 previous: bool = Query(False, description="Get logs from previous terminated container"), 

553) -> Response: 

554 """Get logs from a specific pod belonging to a Job.""" 

555 processor = _check_processor() 

556 _check_namespace(namespace, processor) 

557 

558 try: 

559 pod = processor.core_v1.read_namespaced_pod(name=pod_name, namespace=namespace) 

560 job_name_label = pod.metadata.labels.get("job-name") 

561 if job_name_label != name: 

562 raise HTTPException( 

563 status_code=400, detail=f"Pod '{pod_name}' does not belong to job '{name}'" 

564 ) 

565 

566 log_kwargs: dict[str, Any] = { 

567 "name": pod_name, 

568 "namespace": namespace, 

569 "tail_lines": tail, 

570 "previous": previous, 

571 } 

572 if container: 

573 log_kwargs["container"] = container 

574 

575 log_response = processor.core_v1.read_namespaced_pod_log( 

576 **log_kwargs, _preload_content=False 

577 ) 

578 logs = _decode_pod_log_response(log_response) 

579 

580 response = { 

581 "cluster_id": processor.cluster_id, 

582 "region": processor.region, 

583 "timestamp": datetime.now(UTC).isoformat(), 

584 "job_name": name, 

585 "namespace": namespace, 

586 "pod_name": pod_name, 

587 "container": container, 

588 "logs": logs, 

589 } 

590 

591 return JSONResponse(status_code=200, content=response) 

592 

593 except HTTPException: 

594 raise 

595 except Exception as e: 

596 if "NotFound" in str(e) or "404" in str(e): 

597 raise HTTPException(status_code=404, detail=f"Pod '{pod_name}' not found") from e 

598 raise internal_server_error("getting pod logs", e) from e 

599 

600 

601@router.get("/{namespace}/{name}/metrics") 

602async def get_job_metrics(namespace: str, name: str) -> Response: 

603 """Get resource usage metrics for a Job's pods.""" 

604 processor = _check_processor() 

605 _check_namespace(namespace, processor) 

606 

607 try: 

608 pods = processor.core_v1.list_namespaced_pod( 

609 namespace=namespace, label_selector=f"job-name={name}" 

610 ) 

611 

612 if not pods.items: 

613 raise HTTPException(status_code=404, detail=f"No pods found for job '{name}'") 

614 

615 pod_metrics = [] 

616 total_cpu_millicores = 0 

617 total_memory_bytes = 0 

618 

619 try: 

620 for pod in pods.items: 

621 try: 

622 metrics = processor.custom_objects.get_namespaced_custom_object( 

623 group=METRICS_API_GROUP, 

624 version=METRICS_API_VERSION, 

625 namespace=namespace, 

626 plural="pods", 

627 name=pod.metadata.name, 

628 ) 

629 

630 containers_metrics = [] 

631 for container in metrics.get("containers", []): 

632 cpu_str = container.get("usage", {}).get("cpu", "0") 

633 memory_str = container.get("usage", {}).get("memory", "0") 

634 

635 cpu_millicores = 0 

636 if cpu_str.endswith("n"): 

637 cpu_millicores = int(cpu_str[:-1]) // 1000000 

638 elif cpu_str.endswith("m"): 

639 cpu_millicores = int(cpu_str[:-1]) 

640 else: 

641 cpu_millicores = int(cpu_str) * 1000 

642 

643 memory_bytes = 0 

644 if memory_str.endswith("Ki"): 

645 memory_bytes = int(memory_str[:-2]) * 1024 

646 elif memory_str.endswith("Mi"): 

647 memory_bytes = int(memory_str[:-2]) * 1024 * 1024 

648 elif memory_str.endswith("Gi"): 

649 memory_bytes = int(memory_str[:-2]) * 1024 * 1024 * 1024 

650 else: 

651 memory_bytes = int(memory_str) 

652 

653 total_cpu_millicores += cpu_millicores 

654 total_memory_bytes += memory_bytes 

655 

656 containers_metrics.append( 

657 { 

658 "name": container.get("name"), 

659 "cpu_millicores": cpu_millicores, 

660 "memory_bytes": memory_bytes, 

661 "memory_mib": round(memory_bytes / (1024 * 1024), 2), 

662 } 

663 ) 

664 

665 pod_metrics.append( 

666 {"pod_name": pod.metadata.name, "containers": containers_metrics} 

667 ) 

668 

669 except Exception as e: 

670 logger.warning(f"Could not get metrics for pod {pod.metadata.name}: {e}") 

671 pod_metrics.append( 

672 {"pod_name": pod.metadata.name, "error": "Metrics not available"} 

673 ) 

674 

675 except Exception as e: 

676 logger.warning(f"Metrics API not available: {e}") 

677 

678 response = { 

679 "cluster_id": processor.cluster_id, 

680 "region": processor.region, 

681 "timestamp": datetime.now(UTC).isoformat(), 

682 "job_name": name, 

683 "namespace": namespace, 

684 "summary": { 

685 "total_cpu_millicores": total_cpu_millicores, 

686 "total_memory_bytes": total_memory_bytes, 

687 "total_memory_mib": round(total_memory_bytes / (1024 * 1024), 2), 

688 "pod_count": len(pods.items), 

689 }, 

690 "pods": pod_metrics, 

691 } 

692 

693 return JSONResponse(status_code=200, content=response) 

694 

695 except HTTPException: 

696 raise 

697 except Exception as e: 

698 raise internal_server_error("getting job metrics", e) from e 

699 

700 

701@router.delete("/{namespace}/{name}") 

702async def delete_job( 

703 namespace: str, 

704 name: str, 

705 expected_uid: str | None = Query( 

706 None, 

707 min_length=1, 

708 max_length=128, 

709 description="Optional immutable Kubernetes UID deletion precondition", 

710 ), 

711) -> Response: 

712 """Delete a Job, optionally requiring its immutable Kubernetes UID.""" 

713 processor = _check_processor() 

714 _check_namespace(namespace, processor) 

715 

716 try: 

717 delete_kwargs: dict[str, Any] = { 

718 "name": name, 

719 "namespace": namespace, 

720 } 

721 deleted_uid: str | None = None 

722 if expected_uid is not None: 

723 current = processor.batch_v1.read_namespaced_job(name=name, namespace=namespace) 

724 deleted_uid = str(getattr(current.metadata, "uid", "") or "") 

725 if deleted_uid != expected_uid: 

726 raise HTTPException( 

727 status_code=409, 

728 detail=( 

729 f"Job '{name}' UID changed; expected {expected_uid!r}, " 

730 f"found {deleted_uid or 'unknown'!r}" 

731 ), 

732 ) 

733 delete_kwargs["body"] = kubernetes_client.V1DeleteOptions( 

734 propagation_policy="Background", 

735 preconditions=kubernetes_client.V1Preconditions(uid=expected_uid), 

736 ) 

737 else: 

738 delete_kwargs["propagation_policy"] = "Background" 

739 processor.batch_v1.delete_namespaced_job(**delete_kwargs) 

740 

741 response = { 

742 "cluster_id": processor.cluster_id, 

743 "region": processor.region, 

744 "timestamp": datetime.now(UTC).isoformat(), 

745 "job_name": name, 

746 "namespace": namespace, 

747 "uid": deleted_uid, 

748 "status": "deleted", 

749 "message": "Job deleted successfully", 

750 } 

751 

752 return JSONResponse(status_code=200, content=response) 

753 

754 except HTTPException: 

755 raise 

756 except Exception as e: 

757 status = getattr(e, "status", None) 

758 if status == 409: 

759 raise HTTPException( 

760 status_code=409, 

761 detail=f"Job '{name}' changed before its UID-preconditioned delete", 

762 ) from e 

763 if status == 404 or "NotFound" in str(e) or "404" in str(e): 

764 raise HTTPException( 

765 status_code=404, detail=f"Job '{name}' not found in namespace '{namespace}'" 

766 ) from e