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

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

2 

3from __future__ import annotations 

4 

5import logging 

6import re 

7from datetime import UTC, datetime, timedelta 

8from typing import Any 

9 

10from fastapi import APIRouter, HTTPException, Query 

11from fastapi.responses import JSONResponse, Response 

12from kubernetes import client as kubernetes_client 

13 

14from gco.models import ManifestSubmissionRequest 

15from gco.services.api_shared import ( 

16 BulkDeleteRequest, 

17 _check_namespace, 

18 _check_processor, 

19 _parse_event_to_dict, 

20 _parse_job_to_dict, 

21 _parse_pod_to_dict, 

22) 

23 

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

25logger = logging.getLogger(__name__) 

26 

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

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

29 

30 

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

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

33 if selector is None: 

34 return [] 

35 

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

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

38 clause = raw_clause.strip() 

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

40 raise ValueError( 

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

42 ) 

43 

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

45 if "/" in key: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

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

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

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

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

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

51 ) 

52 else: 

53 name = key 

54 valid_prefix = True 

55 

56 if not valid_prefix or not _LABEL_NAME_RE.fullmatch(name): 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

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

58 if value and not _LABEL_NAME_RE.fullmatch(value): 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

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

60 requirements.append((key, value)) 

61 

62 return requirements 

63 

64 

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

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

67 if not isinstance(labels, dict): 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true

68 return False 

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

70 

71 

72@router.get("") 

73async def list_jobs( 

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

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

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

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

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

79 label_selector: str | None = Query( 

80 None, 

81 max_length=1024, 

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

83 ), 

84) -> Response: 

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

86 processor = _check_processor() 

87 

88 try: 

89 selector_requirements = _parse_exact_label_selector(label_selector) 

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

91 

92 if selector_requirements: 

93 all_jobs = [ 

94 job 

95 for job in all_jobs 

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

97 ] 

98 

99 sort_field, sort_order = "createdAt", "desc" 

100 if ":" in sort: 100 ↛ 103line 100 didn't jump to line 103 because the condition on line 100 was always true

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

102 

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

104 if sort_field == "createdAt": 

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

106 if sort_field == "name": 106 ↛ 108line 106 didn't jump to line 108 because the condition on line 106 was always true

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

108 if sort_field == "status": 

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

110 return "" 

111 

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

113 

114 total = len(all_jobs) 

115 paginated_jobs = all_jobs[offset : offset + limit] 

116 

117 response = { 

118 "cluster_id": processor.cluster_id, 

119 "region": processor.region, 

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

121 "total": total, 

122 "limit": limit, 

123 "offset": offset, 

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

125 "count": len(paginated_jobs), 

126 "jobs": paginated_jobs, 

127 } 

128 

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

130 

131 except ValueError as e: 

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

133 except Exception as e: 

134 logger.error(f"Error listing jobs: {e}") 

135 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

136 

137 

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

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

140 """Get details of a specific Job.""" 

141 processor = _check_processor() 

142 _check_namespace(namespace, processor) 

143 

144 try: 

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

146 job_info = _parse_job_to_dict(job) 

147 

148 response = { 

149 "cluster_id": processor.cluster_id, 

150 "region": processor.region, 

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

152 **job_info, 

153 } 

154 

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

156 

157 except HTTPException: 

158 raise 

159 except Exception as e: 

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

161 raise HTTPException( 

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

163 ) from e 

164 logger.error(f"Error getting job: {e}") 

165 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

166 

167 

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

169async def get_job_logs( 

170 namespace: str, 

171 name: str, 

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

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

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

175 since_seconds: int | None = Query( 

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

177 ), 

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

179) -> Response: 

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

181 from kubernetes.client.rest import ApiException as K8sApiException 

182 

183 processor = _check_processor() 

184 _check_namespace(namespace, processor) 

185 

186 try: 

187 try: 

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

189 except K8sApiException as e: 

190 if e.status == 404: 

191 raise HTTPException( 

192 status_code=404, 

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

194 ) from e 

195 raise 

196 

197 pods = processor.core_v1.list_namespaced_pod( 

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

199 ) 

200 

201 if not pods.items: 

202 raise HTTPException( 

203 status_code=404, 

204 detail=( 

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

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

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

208 ), 

209 ) 

210 

211 sorted_pods = sorted( 

212 pods.items, 

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

214 reverse=True, 

215 ) 

216 pod = sorted_pods[0] 

217 

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

219 if pod_phase == "Pending": 

220 raise HTTPException( 

221 status_code=409, 

222 detail=( 

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

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

225 ), 

226 ) 

227 

228 log_kwargs: dict[str, Any] = { 

229 "name": pod.metadata.name, 

230 "namespace": namespace, 

231 "tail_lines": tail, 

232 "previous": previous, 

233 "timestamps": timestamps, 

234 } 

235 if container: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true

236 log_kwargs["container"] = container 

237 if since_seconds: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

238 log_kwargs["since_seconds"] = since_seconds 

239 

240 try: 

241 logs = processor.core_v1.read_namespaced_pod_log(**log_kwargs) 

242 except K8sApiException as e: 

243 if e.status == 400: 243 ↛ 256line 243 didn't jump to line 256 because the condition on line 243 was always true

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

245 if "waiting" in error_body.lower() or "not found" in error_body.lower(): 245 ↛ 255line 245 didn't jump to line 255 because the condition on line 245 was always true

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

247 raise HTTPException( 

248 status_code=400, 

249 detail=( 

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

251 f"Pod phase: {pod_phase}. " 

252 f"Available containers: {available}" 

253 ), 

254 ) from e 

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

256 raise 

257 

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

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

260 

261 response = { 

262 "cluster_id": processor.cluster_id, 

263 "region": processor.region, 

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

265 "job_name": name, 

266 "namespace": namespace, 

267 "pod_name": pod.metadata.name, 

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

269 "available_containers": available_containers, 

270 "init_containers": init_containers, 

271 "previous": previous, 

272 "tail_lines": tail, 

273 "logs": logs, 

274 } 

275 

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

277 

278 except HTTPException: 

279 raise 

280 except K8sApiException as e: 

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

282 raise HTTPException( 

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

284 ) from e 

285 except Exception as e: 

286 logger.error(f"Error getting job logs: {e}") 

287 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

288 

289 

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

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

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

293 processor = _check_processor() 

294 _check_namespace(namespace, processor) 

295 

296 try: 

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

298 job_events = processor.core_v1.list_namespaced_event( 

299 namespace=namespace, field_selector=field_selector 

300 ) 

301 

302 pods = processor.core_v1.list_namespaced_pod( 

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

304 ) 

305 

306 pod_events = [] 

307 for pod in pods.items: 307 ↛ 308line 307 didn't jump to line 308 because the loop on line 307 never started

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

309 events = processor.core_v1.list_namespaced_event( 

310 namespace=namespace, field_selector=field_selector 

311 ) 

312 pod_events.extend(events.items) 

313 

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

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

316 all_events.sort( 

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

318 ) 

319 

320 response = { 

321 "cluster_id": processor.cluster_id, 

322 "region": processor.region, 

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

324 "job_name": name, 

325 "namespace": namespace, 

326 "count": len(all_events), 

327 "events": all_events, 

328 } 

329 

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

331 

332 except HTTPException: 

333 raise 

334 except Exception as e: 

335 logger.error(f"Error getting job events: {e}") 

336 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

337 

338 

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

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

341 """Get pods belonging to a Job.""" 

342 processor = _check_processor() 

343 _check_namespace(namespace, processor) 

344 

345 try: 

346 pods = processor.core_v1.list_namespaced_pod( 

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

348 ) 

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

350 

351 response = { 

352 "cluster_id": processor.cluster_id, 

353 "region": processor.region, 

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

355 "job_name": name, 

356 "namespace": namespace, 

357 "count": len(pod_list), 

358 "pods": pod_list, 

359 } 

360 

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

362 

363 except HTTPException: 

364 raise 

365 except Exception as e: 

366 logger.error(f"Error getting job pods: {e}") 

367 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

368 

369 

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

371async def get_pod_logs( 

372 namespace: str, 

373 name: str, 

374 pod_name: str, 

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

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

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

378) -> Response: 

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

380 processor = _check_processor() 

381 _check_namespace(namespace, processor) 

382 

383 try: 

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

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

386 if job_name_label != name: 

387 raise HTTPException( 

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

389 ) 

390 

391 log_kwargs: dict[str, Any] = { 

392 "name": pod_name, 

393 "namespace": namespace, 

394 "tail_lines": tail, 

395 "previous": previous, 

396 } 

397 if container: 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true

398 log_kwargs["container"] = container 

399 

400 logs = processor.core_v1.read_namespaced_pod_log(**log_kwargs) 

401 

402 response = { 

403 "cluster_id": processor.cluster_id, 

404 "region": processor.region, 

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

406 "job_name": name, 

407 "namespace": namespace, 

408 "pod_name": pod_name, 

409 "container": container, 

410 "logs": logs, 

411 } 

412 

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

414 

415 except HTTPException: 

416 raise 

417 except Exception as e: 

418 if "NotFound" in str(e) or "404" in str(e): 418 ↛ 420line 418 didn't jump to line 420 because the condition on line 418 was always true

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

420 logger.error(f"Error getting pod logs: {e}") 

421 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

422 

423 

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

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

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

427 processor = _check_processor() 

428 _check_namespace(namespace, processor) 

429 

430 try: 

431 pods = processor.core_v1.list_namespaced_pod( 

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

433 ) 

434 

435 if not pods.items: 

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

437 

438 pod_metrics = [] 

439 total_cpu_millicores = 0 

440 total_memory_bytes = 0 

441 

442 try: 

443 for pod in pods.items: 

444 try: 

445 metrics = processor.custom_objects.get_namespaced_custom_object( 

446 group="metrics.k8s.io", 

447 version="v1beta1", 

448 namespace=namespace, 

449 plural="pods", 

450 name=pod.metadata.name, 

451 ) 

452 

453 containers_metrics = [] 

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

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

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

457 

458 cpu_millicores = 0 

459 if cpu_str.endswith("n"): 

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

461 elif cpu_str.endswith("m"): 

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

463 else: 

464 cpu_millicores = int(cpu_str) * 1000 

465 

466 memory_bytes = 0 

467 if memory_str.endswith("Ki"): 

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

469 elif memory_str.endswith("Mi"): 

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

471 elif memory_str.endswith("Gi"): 471 ↛ 474line 471 didn't jump to line 474 because the condition on line 471 was always true

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

473 else: 

474 memory_bytes = int(memory_str) 

475 

476 total_cpu_millicores += cpu_millicores 

477 total_memory_bytes += memory_bytes 

478 

479 containers_metrics.append( 

480 { 

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

482 "cpu_millicores": cpu_millicores, 

483 "memory_bytes": memory_bytes, 

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

485 } 

486 ) 

487 

488 pod_metrics.append( 

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

490 ) 

491 

492 except Exception as e: 

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

494 pod_metrics.append( 

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

496 ) 

497 

498 except Exception as e: 

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

500 

501 response = { 

502 "cluster_id": processor.cluster_id, 

503 "region": processor.region, 

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

505 "job_name": name, 

506 "namespace": namespace, 

507 "summary": { 

508 "total_cpu_millicores": total_cpu_millicores, 

509 "total_memory_bytes": total_memory_bytes, 

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

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

512 }, 

513 "pods": pod_metrics, 

514 } 

515 

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

517 

518 except HTTPException: 

519 raise 

520 except Exception as e: 

521 logger.error(f"Error getting job metrics: {e}") 

522 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

523 

524 

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

526async def delete_job( 

527 namespace: str, 

528 name: str, 

529 expected_uid: str | None = Query( 

530 None, 

531 min_length=1, 

532 max_length=128, 

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

534 ), 

535) -> Response: 

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

537 processor = _check_processor() 

538 _check_namespace(namespace, processor) 

539 

540 try: 

541 delete_kwargs: dict[str, Any] = { 

542 "name": name, 

543 "namespace": namespace, 

544 } 

545 deleted_uid: str | None = None 

546 if expected_uid is not None: 

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

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

549 if deleted_uid != expected_uid: 

550 raise HTTPException( 

551 status_code=409, 

552 detail=( 

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

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

555 ), 

556 ) 

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

558 propagation_policy="Background", 

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

560 ) 

561 else: 

562 delete_kwargs["propagation_policy"] = "Background" 

563 processor.batch_v1.delete_namespaced_job(**delete_kwargs) 

564 

565 response = { 

566 "cluster_id": processor.cluster_id, 

567 "region": processor.region, 

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

569 "job_name": name, 

570 "namespace": namespace, 

571 "uid": deleted_uid, 

572 "status": "deleted", 

573 "message": "Job deleted successfully", 

574 } 

575 

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

577 

578 except HTTPException: 

579 raise 

580 except Exception as e: 

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

582 if status == 409: 

583 raise HTTPException( 

584 status_code=409, 

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

586 ) from e 

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

588 raise HTTPException( 

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

590 ) from e 

591 logger.error(f"Error deleting job: {e}") 

592 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

593 

594 

595@router.delete("") 

596async def bulk_delete_jobs(request: BulkDeleteRequest) -> Response: 

597 """Bulk delete jobs based on filters.""" 

598 

599 processor = _check_processor() 

600 

601 try: 

602 selector_requirements = _parse_exact_label_selector(request.label_selector) 

603 status_filter = request.status.value if request.status else None 

604 all_jobs = await processor.list_jobs( 

605 namespace=request.namespace, status_filter=status_filter 

606 ) 

607 

608 jobs_to_delete = [] 

609 cutoff_time = None 

610 if request.older_than_days: 

611 cutoff_time = datetime.now(UTC) - timedelta(days=request.older_than_days) 

612 

613 for job in all_jobs: 

614 if cutoff_time: 

615 created_str = job.get("metadata", {}).get("creationTimestamp") 

616 if created_str: 616 ↛ 628line 616 didn't jump to line 628 because the condition on line 616 was always true

617 created = datetime.fromisoformat(created_str.replace("Z", "+00:00")) 

618 # Kubernetes normally returns an aware RFC3339 timestamp, 

619 # but older fixtures/clients may provide a naive value. 

620 # Normalize both forms to aware UTC before comparison. 

621 if created.tzinfo is None: 

622 created = created.replace(tzinfo=UTC) 

623 else: 

624 created = created.astimezone(UTC) 

625 if created > cutoff_time: 

626 continue 

627 

628 if selector_requirements and not _labels_match( 

629 job.get("metadata", {}).get("labels", {}), selector_requirements 

630 ): 

631 continue 

632 

633 jobs_to_delete.append(job) 

634 

635 deleted_jobs = [] 

636 failed_jobs = [] 

637 

638 if not request.dry_run: 

639 for job in jobs_to_delete: 

640 job_name = job.get("metadata", {}).get("name") 

641 job_namespace = job.get("metadata", {}).get("namespace") 

642 try: 

643 processor.batch_v1.delete_namespaced_job( 

644 name=job_name, namespace=job_namespace, propagation_policy="Background" 

645 ) 

646 deleted_jobs.append({"name": job_name, "namespace": job_namespace}) 

647 except Exception as e: 

648 failed_jobs.append( 

649 {"name": job_name, "namespace": job_namespace, "error": str(e)} 

650 ) 

651 

652 response: dict[str, Any] = { 

653 "cluster_id": processor.cluster_id, 

654 "region": processor.region, 

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

656 "dry_run": request.dry_run, 

657 "total_matched": len(jobs_to_delete), 

658 "deleted_count": len(deleted_jobs), 

659 "failed_count": len(failed_jobs), 

660 "jobs": ( 

661 [ 

662 { 

663 "name": j.get("metadata", {}).get("name"), 

664 "namespace": j.get("metadata", {}).get("namespace"), 

665 } 

666 for j in jobs_to_delete 

667 ] 

668 if request.dry_run 

669 else deleted_jobs 

670 ), 

671 "failed": failed_jobs if failed_jobs else None, 

672 } 

673 

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

675 

676 except ValueError as e: 

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

678 except Exception as e: 

679 logger.error(f"Error bulk deleting jobs: {e}") 

680 raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}") from e 

681 

682 

683@router.post("/{namespace}/{name}/retry") 

684async def retry_job(namespace: str, name: str) -> Response: 

685 """Retry a failed job by creating a new job from its spec.""" 

686 processor = _check_processor() 

687 _check_namespace(namespace, processor) 

688 

689 try: 

690 try: 

691 original_job = processor.batch_v1.read_namespaced_job(name=name, namespace=namespace) 

692 except Exception as e: 

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

694 raise HTTPException( 

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

696 ) from e 

697 raise 

698 

699 new_name = f"{name}-retry-{datetime.now(UTC).strftime('%Y%m%d%H%M%S')}" 

700 

701 new_job_manifest = { 

702 "apiVersion": "batch/v1", 

703 "kind": "Job", 

704 "metadata": { 

705 "name": new_name, 

706 "namespace": namespace, 

707 "labels": { 

708 **(original_job.metadata.labels or {}), 

709 "gco.io/retry-of": name, 

710 }, 

711 "annotations": { 

712 **(original_job.metadata.annotations or {}), 

713 "gco.io/original-job": name, 

714 }, 

715 }, 

716 "spec": { 

717 "parallelism": original_job.spec.parallelism, 

718 "completions": original_job.spec.completions, 

719 "backoffLimit": original_job.spec.backoff_limit, 

720 "template": original_job.spec.template.to_dict(), 

721 }, 

722 } 

723 

724 spec_dict = new_job_manifest.get("spec", {}) 

725 if isinstance(spec_dict, dict): 725 ↛ 730line 725 didn't jump to line 730 because the condition on line 725 was always true

726 template_dict = spec_dict.get("template", {}) 

727 if isinstance(template_dict, dict) and "status" in template_dict: 727 ↛ 728line 727 didn't jump to line 728 because the condition on line 727 was never true

728 del template_dict["status"] 

729 

730 submission_request = ManifestSubmissionRequest( 

731 manifests=[new_job_manifest], namespace=namespace, dry_run=False, validate=True 

732 ) 

733 

734 result = await processor.process_manifest_submission(submission_request) 

735 

736 response = {