Coverage for cli/jobs.py: 94%

1""" 

2Job management for GCO CLI. 

3 

4Provides functionality to submit, query, and manage jobs across GCO clusters. 

5""" 

6 

7from collections.abc import Callable 

8from dataclasses import dataclass, field 

9from datetime import UTC, datetime 

10from pathlib import Path 

11from typing import Any 

12 

13import yaml 

14 

15from gco.services.manifest_processor import safe_load_all_yaml 

16 

17from .aws_client import get_aws_client 

18from .config import GCOConfig, get_config 

19 

20# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

21# Flowchart(s) generated from this file: 

22# * ``JobManager.submit_job`` -> ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job.html`` 

23# (PNG: ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job.png``) 

24# * ``JobManager.submit_job_sqs`` -> ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job_sqs.html`` 

25# (PNG: ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job_sqs.png``) 

26# Regenerate with ``python diagrams/code_diagrams/generate.py``. 

27# <pyflowchart-code-diagram> END 

28 

29 

30logger = __import__("logging").getLogger(__name__) 

31 

32 

33def _format_duration(seconds: int) -> str: 

34 """Format seconds into a human-readable duration string.""" 

35 if seconds < 60: 

36 return f"{seconds}s" 

37 minutes, secs = divmod(seconds, 60) 

38 if minutes < 60: 

39 return f"{minutes}m{secs:02d}s" 

40 hours, mins = divmod(minutes, 60) 

41 return f"{hours}h{mins:02d}m{secs:02d}s" 

42 

43 

44def _first_manifest_namespace(manifests: list[dict[str, Any]]) -> str | None: 

45 """Return the first explicit ``metadata.namespace`` found in a manifest list. 

46 

47 Used by the SQS submission path to populate the envelope ``namespace`` 

48 field (informational — the queue processor reads each manifest's own 

49 namespace for validation). Returns None if no manifest declares one. 

50 """ 

51 for manifest in manifests: 

52 ns = manifest.get("metadata", {}).get("namespace") if isinstance(manifest, dict) else None 

53 if ns: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 return str(ns) 

55 return None 

56 

57 

58def _extract_image_refs(spec: dict[str, Any]) -> list[str]: 

59 """Extract container image refs from a parsed Job spec. 

60 

61 The API surface for a Job carries ``spec.template.spec.containers`` and 

62 ``spec.template.spec.initContainers`` lists, each entry of which has 

63 a ``name`` and an ``image`` URI. Returns an alphabetically-sorted, 

64 deduplicated list so the output is stable across calls — orphan-image 

65 cross-references rely on set equality. 

66 """ 

67 refs: set[str] = set() 

68 template = spec.get("template") if isinstance(spec, dict) else None 

69 pod_spec = template.get("spec") if isinstance(template, dict) else None 

70 if not isinstance(pod_spec, dict): 

71 return [] 

72 for key in ("containers", "initContainers"): 

73 items = pod_spec.get(key, []) 

74 if not isinstance(items, list): 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 continue 

76 for entry in items: 

77 if not isinstance(entry, dict): 

78 continue 

79 image = entry.get("image") 

80 if isinstance(image, str) and image: 

81 refs.add(image) 

82 return sorted(refs) 

83 

84 

85@dataclass 

86class JobInfo: 

87 """Information about a Kubernetes job.""" 

88 

89 name: str 

90 namespace: str 

91 region: str 

92 status: str # "pending", "running", "succeeded", "failed" 

93 created_time: datetime | None = None 

94 start_time: datetime | None = None 

95 completion_time: datetime | None = None 

96 active_pods: int = 0 

97 succeeded_pods: int = 0 

98 failed_pods: int = 0 

99 parallelism: int = 1 

100 completions: int = 1 

101 labels: dict[str, str] = field(default_factory=dict) 

102 image_refs: list[str] = field(default_factory=list) 

103 

104 @property 

105 def is_complete(self) -> bool: 

106 return self.status in ("succeeded", "failed") 

107 

108 @property 

109 def duration_seconds(self) -> int | None: 

110 if self.start_time and self.completion_time: 

111 return int((self.completion_time - self.start_time).total_seconds()) 

112 if self.start_time: 

113 return int((datetime.now(UTC) - self.start_time).total_seconds()) 

114 return None 

115 

116 

117class JobManager: 

118 """ 

119 Manages jobs across GCO clusters. 

120 

121 Provides: 

122 - Job submission with region targeting 

123 - Job status queries across regions 

124 - Job logs retrieval 

125 - Job deletion 

126 """ 

127 

128 def __init__(self, config: GCOConfig | None = None): 

129 self.config = config or get_config() 

130 self._aws_client = get_aws_client(config) 

131 # Configure regional API mode if enabled 

132 if hasattr(self.config, "use_regional_api") and self.config.use_regional_api: 

133 self._aws_client.set_use_regional_api(True) 

134 

135 def load_manifests(self, path: str) -> list[dict[str, Any]]: 

136 """ 

137 Load Kubernetes manifests from a file or directory. 

138 

139 Args: 

140 path: Path to YAML file or directory containing YAML files 

141 

142 Returns: 

143 List of manifest dictionaries 

144 """ 

145 manifests = [] 

146 path_obj = Path(path) 

147 

148 if path_obj.is_file(): 

149 manifests.extend(self._load_yaml_file(path_obj)) 

150 elif path_obj.is_dir(): 

151 for yaml_file in sorted(path_obj.glob("*.yaml")): 

152 manifests.extend(self._load_yaml_file(yaml_file)) 

153 for yaml_file in sorted(path_obj.glob("*.yml")): 

154 manifests.extend(self._load_yaml_file(yaml_file)) 

155 else: 

156 raise FileNotFoundError(f"Path not found: {path}") 

157 

158 return manifests 

159 

160 def _load_yaml_file(self, path: Path) -> list[dict[str, Any]]: 

161 """Load manifests from a single YAML file.""" 

162 with open(path, encoding="utf-8") as f: 

163 return safe_load_all_yaml(f, allow_aliases=False) 

164 

165 def submit_job( 

166 self, 

167 manifests: str | list[dict[str, Any]], 

168 namespace: str | None = None, 

169 target_region: str | None = None, 

170 dry_run: bool = False, 

171 labels: dict[str, str] | None = None, 

172 ) -> dict[str, Any]: 

173 """ 

174 Submit a job to GCO. 

175 

176 Args: 

177 manifests: Path to manifest file/directory or list of manifest dicts 

178 namespace: Fallback namespace for manifests that don't declare 

179 their own. When set, each manifest's ``metadata.namespace`` is 

180 filled in only if missing — existing values are preserved so 

181 users who've declared a target namespace in the manifest can 

182 rely on it reaching the server untouched. Server-side 

183 validation enforces the allowlist. 

184 target_region: Force job to specific region 

185 dry_run: Validate without applying 

186 labels: Additional labels to add to manifests 

187 

188 Returns: 

189 Submission result dictionary 

190 """ 

191 # Load manifests if path provided 

192 manifest_list = self.load_manifests(manifests) if isinstance(manifests, str) else manifests 

193 

194 # Apply namespace as a fallback only — preserve any namespace the 

195 # manifest declared itself. 

196 if namespace: 

197 for manifest in manifest_list: 

198 if "metadata" not in manifest: 

199 manifest["metadata"] = {} 

200 manifest["metadata"].setdefault("namespace", namespace) 

201 

202 # Apply additional labels 

203 if labels: 

204 for manifest in manifest_list: 

205 if "metadata" not in manifest: 

206 manifest["metadata"] = {} 

207 if "labels" not in manifest["metadata"]: 207 ↛ 209line 207 didn't jump to line 209 because the condition on line 207 was always true

208 manifest["metadata"]["labels"] = {} 

209 manifest["metadata"]["labels"].update(labels) 

210 

211 # Submit via API 

212 return self._aws_client.submit_manifests( 

213 manifests=manifest_list, 

214 namespace=namespace, 

215 target_region=target_region, 

216 dry_run=dry_run, 

217 ) 

218 

219 def submit_job_direct( 

220 self, 

221 manifests: str | list[dict[str, Any]], 

222 region: str, 

223 namespace: str | None = None, 

224 dry_run: bool = False, 

225 labels: dict[str, str] | None = None, 

226 ) -> dict[str, Any]: 

227 """ 

228 Submit a job directly to a regional cluster using kubectl. 

229 

230 This bypasses the API Gateway and submits directly to the EKS cluster 

231 using kubectl. Requires: 

232 - kubectl installed and in PATH 

233 - EKS access entry configured for your IAM principal 

234 - AWS credentials with eks:DescribeCluster permission 

235 

236 Args: 

237 manifests: Path to manifest file/directory or list of manifest dicts 

238 region: Target region for direct submission (required) 

239 namespace: Fallback namespace for manifests that don't declare 

240 their own. When set, each manifest's ``metadata.namespace`` is 

241 filled in only if missing — existing values are preserved so 

242 users who've declared a target namespace in the manifest can 

243 rely on it reaching ``kubectl apply`` untouched. 

244 dry_run: Validate without applying 

245 labels: Additional labels to add to manifests 

246 

247 Returns: 

248 Submission result dictionary 

249 """ 

250 import subprocess 

251 import tempfile 

252 import uuid 

253 

254 # Load manifests if path provided 

255 manifest_list = self.load_manifests(manifests) if isinstance(manifests, str) else manifests 

256 

257 # Apply namespace as a fallback only — preserve any namespace the 

258 # manifest declared itself. 

259 if namespace: 

260 for manifest in manifest_list: 

261 if "metadata" not in manifest: 

262 manifest["metadata"] = {} 

263 manifest["metadata"].setdefault("namespace", namespace) 

264 

265 # Apply additional labels 

266 if labels: 

267 for manifest in manifest_list: 

268 if "metadata" not in manifest: 

269 manifest["metadata"] = {} 

270 if "labels" not in manifest["metadata"]: 270 ↛ 272line 270 didn't jump to line 272 because the condition on line 270 was always true

271 manifest["metadata"]["labels"] = {} 

272 manifest["metadata"]["labels"].update(labels) 

273 

274 # Get cluster name from stack 

275 stack = self._aws_client.get_regional_stack(region) 

276 if not stack: 

277 raise ValueError(f"No GCO stack found in region {region}") 

278 

279 cluster_name = stack.cluster_name 

280 

281 # Update kubeconfig for the cluster 

282 from .kubectl_helpers import update_kubeconfig 

283 

284 update_kubeconfig(cluster_name, region) 

285 

286 # Handle existing Job resources before applying 

287 warnings: list[str] = [] 

288 if not dry_run: 

289 for manifest in manifest_list: 

290 if manifest.get("kind") != "Job": 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true

291 continue 

292 job_name = manifest.get("metadata", {}).get("name") 

293 job_ns = manifest.get("metadata", {}).get("namespace", namespace or "default") 

294 if not job_name: 

295 continue 

296 

297 existing_status = self._get_kubectl_job_status(job_name, job_ns) 

298 if existing_status is None: 

299 # No existing job — nothing to do 

300 continue 

301 

302 if existing_status in ("complete", "failed"): 

303 # Finished job — safe to delete and replace 

304 subprocess.run( 

305 ["kubectl", "delete", "job", job_name, "-n", job_ns], 

306 capture_output=True, 

307 text=True, 

308 ) 

309 else: 

310 # Job is still active — auto-rename to avoid collision 

311 suffix = uuid.uuid4().hex[:5] 

312 new_name = f"{job_name}-{suffix}" 

313 original_name = job_name 

314 manifest["metadata"]["name"] = new_name 

315 warnings.append( 

316 f"Job '{original_name}' is still running in namespace " 

317 f"'{job_ns}'. Renamed new submission to '{new_name}'." 

318 ) 

319 logger.warning( 

320 "Job %s is active in %s, renamed to %s", 

321 original_name, 

322 job_ns, 

323 new_name, 

324 ) 

325 

326 # Write manifests to temp file 

327 with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: 

328 yaml.dump_all(manifest_list, f) 

329 f.flush() # Ensure content is written before using f.name 

330 temp_path = f.name # nosemgrep: tempfile-without-flush 

331 

332 try: 

333 # Build kubectl command 

334 kubectl_cmd = ["kubectl", "apply", "-f", temp_path] 

335 

336 if dry_run: 

337 kubectl_cmd.extend(["--dry-run=client"]) 

338 

339 # Run kubectl apply 

340 result = subprocess.run( 

341 kubectl_cmd, capture_output=True, text=True 

342 ) # nosemgrep: dangerous-subprocess-use-audit - kubectl_cmd is a list ["kubectl","apply","-f",temp_path]; temp_path is a secure tempfile, not user input 

343 

344 if result.returncode != 0: 

345 raise RuntimeError(f"kubectl apply failed: {result.stderr}") 

346 

347 # Parse output to get job name 

348 output_lines = result.stdout.strip().split("\n") 

349 created_resources = [] 

350 for line in output_lines: 

351 if line: 351 ↛ 350line 351 didn't jump to line 350 because the condition on line 351 was always true

352 created_resources.append(line) 

353 

354 # Get job name from first manifest (may have been renamed) 

355 job_name = None 

356 for manifest in manifest_list: 356 ↛ 361line 356 didn't jump to line 361 because the loop on line 356 didn't complete

357 if manifest.get("kind") == "Job": 357 ↛ 356line 357 didn't jump to line 356 because the condition on line 357 was always true

358 job_name = manifest.get("metadata", {}).get("name") 

359 break 

360 

361 response: dict[str, Any] = { 

362 "status": "success", 

363 "method": "kubectl", 

364 "cluster": cluster_name, 

365 "region": region, 

366 "namespace": namespace or "default", 

367 "job_name": job_name, 

368 "dry_run": dry_run, 

369 "resources": created_resources, 

370 "output": result.stdout, 

371 } 

372 if warnings: 

373 response["warnings"] = warnings 

374 return response 

375 

376 finally: 

377 # Clean up temp file 

378 import os 

379 

380 os.unlink(temp_path) 

381 

382 def _get_kubectl_job_status(self, job_name: str, namespace: str) -> str | None: 

383 """Check the status of an existing Job via kubectl. 

384 

385 Returns: 

386 "complete", "failed", "active", or None if the job doesn't exist. 

387 """ 

388 import json 

389 import subprocess 

390 

391 result = subprocess.run( 

392 [ 

393 "kubectl", 

394 "get", 

395 "job", 

396 job_name, 

397 "-n", 

398 namespace, 

399 "-o", 

400 "json", 

401 ], 

402 capture_output=True, 

403 text=True, 

404 ) 

405 if result.returncode != 0: 

406 return None # Job doesn't exist 

407 

408 try: 

409 job_data = json.loads(result.stdout) 

410 except json.JSONDecodeError, KeyError: 

411 return None 

412 

413 conditions = job_data.get("status", {}).get("conditions") or [] 

414 for condition in conditions: 

415 cond_type = condition.get("type", "") 

416 cond_status = condition.get("status", "") 

417 if cond_type == "Complete" and cond_status == "True": 

418 return "complete" 

419 if cond_type == "Failed" and cond_status == "True": 

420 return "failed" 

421 return "active" 

422 

423 def list_jobs( 

424 self, 

425 region: str | None = None, 

426 namespace: str | None = None, 

427 status: str | None = None, 

428 all_regions: bool = False, 

429 ) -> list[JobInfo]: 

430 """ 

431 List jobs across GCO clusters. 

432 

433 Args: 

434 region: Specific region to query 

435 namespace: Filter by namespace 

436 status: Filter by status 

437 all_regions: Query all discovered regions 

438 

439 Returns: 

440 List of JobInfo objects 

441 """ 

442 jobs = [] 

443 

444 if all_regions: 

445 # Query all discovered regional stacks 

446 stacks = self._aws_client.discover_regional_stacks() 

447 for stack_region in stacks: 

448 try: 

449 region_jobs = self._query_jobs_in_region(stack_region, namespace, status) 

450 jobs.extend(region_jobs) 

451 except Exception as e: 

452 logger.warning("Failed to query jobs in %s: %s", stack_region, e) 

453 continue 

454 elif region: 

455 jobs = self._query_jobs_in_region(region, namespace, status) 

456 else: 

457 # Use default region 

458 jobs = self._query_jobs_in_region(self.config.default_region, namespace, status) 

459 

460 return jobs 

461 

462 def _query_jobs_in_region( 

463 self, region: str, namespace: str | None, status: str | None 

464 ) -> list[JobInfo]: 

465 """Query jobs in a specific region.""" 

466 try: 

467 response = self._aws_client.get_jobs(region=region, namespace=namespace, status=status) 

468 

469 jobs = [] 

470 # response is a list, but we expect a dict with "jobs" key from the API 

471 job_list = response.get("jobs", []) if isinstance(response, dict) else response 

472 for job_data in job_list: 

473 jobs.append(self._parse_job_info(job_data, region)) 

474 

475 return jobs 

476 except Exception as exc: 

477 logger.warning("Failed to query jobs in %s: %s", region, exc) 

478 return [] 

479 

480 def _parse_job_info(self, job_data: dict[str, Any], region: str) -> JobInfo: 

481 """Parse job data into JobInfo object.""" 

482 metadata = job_data.get("metadata", {}) 

483 status_data = job_data.get("status", {}) 

484 spec = job_data.get("spec", {}) 

485 

486 # Determine job status 

487 conditions = status_data.get("conditions", []) 

488 job_status = "pending" 

489 for condition in conditions: 

490 if condition.get("type") == "Complete" and condition.get("status") == "True": 

491 job_status = "succeeded" 

492 break 

493 if condition.get("type") == "Failed" and condition.get("status") == "True": 493 ↛ 489line 493 didn't jump to line 489 because the condition on line 493 was always true

494 job_status = "failed" 

495 break 

496 

497 if job_status == "pending" and status_data.get("active", 0) > 0: 

498 job_status = "running" 

499 

500 # Parse timestamps 

501 created_time = None 

502 if metadata.get("creationTimestamp"): 

503 created_time = datetime.fromisoformat( 

504 metadata["creationTimestamp"].replace("Z", "+00:00") 

505 ) 

506 

507 start_time = None 

508 if status_data.get("startTime"): 

509 start_time = datetime.fromisoformat(status_data["startTime"].replace("Z", "+00:00")) 

510 

511 completion_time = None 

512 if status_data.get("completionTime"): 

513 completion_time = datetime.fromisoformat( 

514 status_data["completionTime"].replace("Z", "+00:00") 

515 ) 

516 

517 return JobInfo( 

518 name=metadata.get("name", ""), 

519 namespace=metadata.get("namespace", "default"), 

520 region=region, 

521 status=job_status, 

522 created_time=created_time, 

523 start_time=start_time, 

524 completion_time=completion_time, 

525 active_pods=status_data.get("active", 0), 

526 succeeded_pods=status_data.get("succeeded", 0), 

527 failed_pods=status_data.get("failed", 0), 

528 parallelism=spec.get("parallelism", 1), 

529 completions=spec.get("completions", 1), 

530 labels=metadata.get("labels", {}), 

531 image_refs=_extract_image_refs(spec), 

532 ) 

533 

534 def get_job(self, job_name: str, namespace: str, region: str | None = None) -> JobInfo | None: 

535 """ 

536 Get detailed information about a specific job. 

537 

538 Args: 

539 job_name: Name of the job 

540 namespace: Namespace of the job 

541 region: Region where the job is running 

542 

543 Returns: 

544 JobInfo or None if not found 

545 """ 

546 try: 

547 response = self._aws_client.get_job_details( 

548 job_name=job_name, namespace=namespace, region=region or self.config.default_region 

549 ) 

550 return self._parse_job_info(response, region or self.config.default_region) 

551 except Exception as e: 

552 logger.debug("Failed to get job details for %s: %s", job_name, e) 

553 return None 

554 

555 def get_job_logs( 

556 self, 

557 job_name: str, 

558 namespace: str, 

559 region: str | None = None, 

560 tail_lines: int = 100, 

561 follow: bool = False, 

562 since_hours: int = 24, 

563 ) -> str: 

564 """ 

565 Get logs from a job. 

566 

567 Tries the Kubernetes API first (via the GCO API). If the pod is no 

568 longer available (completed/deleted), falls back to CloudWatch Logs 

569 where Container Insights stores application logs. 

570 

571 Args: 

572 job_name: Name of the job 

573 namespace: Namespace of the job 

574 region: Region where the job is running 

575 tail_lines: Number of lines to return 

576 follow: Stream logs (not implemented yet) 

577 since_hours: Hours to look back in CloudWatch (default 24) 

578 

579 Returns: 

580 Log content as string 

581 """ 

582 if follow: 

583 raise NotImplementedError("Log streaming not yet implemented") 

584 

585 target_region = region or self.config.default_region 

586 

587 try: 

588 return self._aws_client.get_job_logs( 

589 job_name=job_name, 

590 namespace=namespace, 

591 region=target_region, 

592 tail_lines=tail_lines, 

593 ) 

594 except RuntimeError as e: 

595 error_msg = str(e) 

596 # If the pod is gone or pending, try CloudWatch 

597 if any( 

598 hint in error_msg.lower() 

599 for hint in ["not found", "pending", "no pods", "terminated", "completed"] 

600 ): 

601 logger.info("Pod not available, falling back to CloudWatch Logs") 

602 try: 

603 return self._get_cloudwatch_logs( 

604 job_name=job_name, 

605 region=target_region, 

606 tail_lines=tail_lines, 

607 since_hours=since_hours, 

608 ) 

609 except Exception as cw_err: 

610 logger.debug("CloudWatch fallback failed: %s", cw_err) 

611 raise RuntimeError( 

612 f"{error_msg}\n\n" 

613 f"CloudWatch Logs fallback also failed: {cw_err}\n" 

614 f"Tip: Container logs appear in CloudWatch within a few minutes. " 

615 f"If the job just finished, try again shortly." 

616 ) from e 

617 raise 

618 

619 def _get_cloudwatch_logs( 

620 self, 

621 job_name: str, 

622 region: str, 

623 tail_lines: int = 100, 

624 since_hours: int = 24, 

625 ) -> str: 

626 """ 

627 Fetch job logs from CloudWatch Logs (Container Insights). 

628 

629 The CloudWatch Observability addon ships container stdout/stderr to: 

630 /aws/containerinsights/{cluster_name}/application 

631 

632 Args: 

633 job_name: Name of the job (used to filter log streams) 

634 region: AWS region 

635 tail_lines: Number of log lines to return 

636 since_hours: Hours to look back (default 24) 

637 

638 Returns: 

639 Log content as string 

640 """ 

641 cluster_name = f"{self.config.project_name}-{region}" 

642 log_group = f"/aws/containerinsights/{cluster_name}/application" 

643 

644 logs_client = self._aws_client._session.client("logs", region_name=region) 

645 

646 import time 

647 

648 now = int(time.time()) 

649 start_time = now - (since_hours * 3600) 

650 

651 query = ( 

652 f"fields @timestamp, @message " 

653 f'| filter @logStream like "{job_name}" ' 

654 f"| sort @timestamp asc " 

655 f"| limit {tail_lines}" 

656 ) 

657 

658 start_query = logs_client.start_query( 

659 logGroupName=log_group, 

660 startTime=start_time, 

661 endTime=now, 

662 queryString=query, 

663 ) 

664 query_id = start_query["queryId"] 

665 

666 # Poll for results (CloudWatch Insights is async) 

667 result = None 

668 for _ in range(30): # up to 30 seconds 

669 time.sleep(1) 

670 result = logs_client.get_query_results(queryId=query_id) 

671 if result["status"] in ("Complete", "Failed", "Cancelled"): 

672 break 

673 

674 if result is None or result["status"] != "Complete": 

675 status = result["status"] if result else "unknown" 

676 raise RuntimeError( 

677 f"CloudWatch Logs query did not complete (status: {status}). Try again in a moment." 

678 ) 

679 

680 if not result["results"]: 

681 raise RuntimeError( 

682 f"No logs found in CloudWatch for job '{job_name}' " 

683 f"in the last {since_hours} hours (log group: {log_group}). " 

684 f"Logs may take 1-2 minutes to appear after a pod runs. " 

685 f"Use --since to search further back, or check the job name " 

686 f"with: gco jobs list -r {region}" 

687 ) 

688 

689 # Extract log messages from results. 

690 # CloudWatch Container Insights wraps logs in a JSON envelope: 

691 # {"time":"...","stream":"stdout","log":"actual message","kubernetes":{...}} 

692 # We parse out the "log" field for clean output, falling back to the 

693 # raw message if it's not JSON. 

694 import json as _json 

695 

696 lines = [] 

697 for row in result["results"]: 

698 for entry in row: 698 ↛ 697line 698 didn't jump to line 697 because the loop on line 698 didn't complete

699 if entry["field"] == "@message": 

700 raw = entry["value"].rstrip() 

701 try: 

702 parsed = _json.loads(raw) 

703 lines.append(parsed.get("log", raw).rstrip()) 

704 except ValueError, TypeError: 

705 lines.append(raw) 

706 break 

707 

708 header = f"[CloudWatch Logs — {log_group}]\n" 

709 return header + "\n".join(lines) 

710 

711 def delete_job( 

712 self, job_name: str, namespace: str, region: str | None = None 

713 ) -> dict[str, Any]: 

714 """ 

715 Delete a job. 

716 

717 Args: 

718 job_name: Name of the job 

719 namespace: Namespace of the job 

720 region: Region where the job is running 

721 

722 Returns: 

723 Deletion result 

724 """ 

725 return self._aws_client.delete_job( 

726 job_name=job_name, namespace=namespace, region=region or self.config.default_region 

727 ) 

728 

729 def wait_for_job( 

730 self, 

731 job_name: str, 

732 namespace: str, 

733 region: str | None = None, 

734 timeout_seconds: int = 3600, 

735 poll_interval: int = 10, 

736 progress_callback: Callable[[JobInfo, int], None] | None = None, 

737 ) -> JobInfo: 

738 """ 

739 Wait for a job to complete with progress reporting. 

740 

741 Args: 

742 job_name: Name of the job 

743 namespace: Namespace of the job 

744 region: Region where the job is running 

745 timeout_seconds: Maximum time to wait 

746 poll_interval: Seconds between status checks 

747 progress_callback: Optional callable(JobInfo, elapsed_seconds) for progress updates. 

748 If None, a default stderr progress line is printed. 

749 

750 Returns: 

751 Final JobInfo 

752 

753 Raises: 

754 TimeoutError: If job doesn't complete within timeout 

755 """ 

756 import sys 

757 import time 

758 

759 start_time = time.time() 

760 

761 while True: 

762 job = self.get_job(job_name, namespace, region) 

763 

764 if job is None: 

765 raise ValueError(f"Job {job_name} not found in namespace {namespace}") 

766 

767 elapsed = time.time() - start_time 

768 elapsed_str = _format_duration(int(elapsed)) 

769 

770 if job.is_complete: 

771 # Clear the progress line and return 

772 sys.stderr.write("\r\033[K") 

773 sys.stderr.flush() 

774 return job 

775 

776 # Build progress message 

777 pods_info = ( 

778 f"{job.active_pods} active, {job.succeeded_pods}/{job.completions} succeeded" 

779 ) 

780 if job.failed_pods: 780 ↛ 781line 780 didn't jump to line 781 because the condition on line 780 was never true

781 pods_info += f", {job.failed_pods} failed" 

782 

783 status_line = f" ⏳ {job.status.capitalize()} — {pods_info} — {elapsed_str} elapsed" 

784 

785 if progress_callback: 785 ↛ 789line 785 didn't jump to line 789 because the condition on line 785 was always true

786 progress_callback(job, int(elapsed)) 

787 else: 

788 # Overwrite the same line on stderr 

789 sys.stderr.write(f"\r\033[K{status_line}") 

790 sys.stderr.flush() 

791 

792 if elapsed >= timeout_seconds: 

793 sys.stderr.write("\r\033[K") 

794 sys.stderr.flush() 

795 raise TimeoutError( 

796 f"Job {job_name} did not complete within {timeout_seconds} seconds " 

797 f"(last status: {job.status}, pods: {pods_info})" 

798 ) 

799 

800 time.sleep(poll_interval) # nosemgrep: arbitrary-sleep - intentional polling delay 

801 

802 def submit_job_sqs( 

803 self, 

804 manifests: str | list[dict[str, Any]], 

805 region: str, 

806 namespace: str | None = None, 

807 labels: dict[str, str] | None = None, 

808 priority: int = 0, 

809 ) -> dict[str, Any]: 

810 """ 

811 Submit a job to a regional SQS queue for processing. 

812 

813 This is the recommended way to submit jobs as it: 

814 - Decouples submission from processing 

815 - Enables KEDA-based autoscaling 

816 - Provides better fault tolerance 

817 

818 Args: 

819 manifests: Path to manifest file/directory or list of manifest dicts 

820 region: Target region for job submission (required) 

821 namespace: Fallback namespace for manifests that don't declare 

822 their own. When set, each manifest's ``metadata.namespace`` is 

823 filled in only if missing — existing values are preserved so 

824 users who've declared a target namespace in the manifest can 

825 rely on it reaching the queue processor untouched. Server-side 

826 validation enforces the allowlist. 

827 labels: Additional labels to add to manifests 

828 priority: Job priority (higher = more important) 

829 

830 Returns: 

831 Submission result dictionary with message_id and queue info 

832 """ 

833 import json 

834 import uuid 

835 

836 import boto3 

837 

838 # Load manifests if path provided 

839 manifest_list = self.load_manifests(manifests) if isinstance(manifests, str) else manifests 

840 

841 # Apply namespace as a fallback only — preserve any namespace the 

842 # manifest declared itself. 

843 if namespace: 

844 for manifest in manifest_list: 

845 if "metadata" not in manifest: 

846 manifest["metadata"] = {} 

847 manifest["metadata"].setdefault("namespace", namespace) 

848 

849 # Apply additional labels 

850 if labels: 

851 for manifest in manifest_list: 

852 if "metadata" not in manifest: 852 ↛ 853line 852 didn't jump to line 853 because the condition on line 852 was never true

853 manifest["metadata"] = {} 

854 if "labels" not in manifest["metadata"]: 854 ↛ 856line 854 didn't jump to line 856 because the condition on line 854 was always true

855 manifest["metadata"]["labels"] = {} 

856 manifest["metadata"]["labels"].update(labels) 

857 

858 # Get queue URL from stack 

859 stack = self._aws_client.get_regional_stack(region)