Coverage for cli/jobs.py: 92.75%

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, Mapping 

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# Generated at (UTC): 2026-07-18T01:03:40Z 

22# Flowchart(s) generated from this file: 

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

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

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

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

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

28# <pyflowchart-code-diagram> END 

29 

30 

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

32 

33 

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

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

36 if seconds < 60: 

37 return f"{seconds}s" 

38 minutes, secs = divmod(seconds, 60) 

39 if minutes < 60: 

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

41 hours, mins = divmod(minutes, 60) 

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

43 

44 

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

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

47 

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

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

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

51 """ 

52 for manifest in manifests: 

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

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

55 return str(ns) 

56 return None 

57 

58 

59def resolve_submission_identity( 

60 result: Any, 

61 *, 

62 fallback_name: str | None = None, 

63 fallback_namespace: str | None = None, 

64) -> tuple[str | None, str | None]: 

65 """Resolve the submitted Job name and namespace from supported responses. 

66 

67 API submissions return resource-status dictionaries, while direct kubectl 

68 submissions return a top-level ``job_name`` and a ``resources`` list of 

69 human-readable strings. Only mapping-shaped resources are inspected, so 

70 direct response lines can never be mistaken for response envelopes. 

71 """ 

72 if not isinstance(result, Mapping): 

73 return fallback_name, fallback_namespace 

74 

75 raw_resources = result.get("resources") or [] 

76 if isinstance(raw_resources, Mapping): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 raw_resources = [raw_resources] 

78 resources = [resource for resource in raw_resources if isinstance(resource, Mapping)] 

79 job_resources = [ 

80 resource for resource in resources if str(resource.get("kind", "")).lower() == "job" 

81 ] 

82 

83 explicit_job_name = result.get("job_name") 

84 resource_with_name = next( 

85 (resource for resource in job_resources if resource.get("name")), None 

86 ) 

87 job_name = ( 

88 str(explicit_job_name) 

89 if explicit_job_name 

90 else str(resource_with_name.get("name")) 

91 if resource_with_name is not None 

92 else fallback_name 

93 ) 

94 

95 matching_resource = next( 

96 (resource for resource in job_resources if resource.get("name") == job_name), 

97 resource_with_name, 

98 ) 

99 resource_namespace = matching_resource.get("namespace") if matching_resource else None 

100 envelope_namespace = result.get("namespace") 

101 namespace = ( 

102 str(resource_namespace) 

103 if resource_namespace 

104 else str(envelope_namespace) 

105 if envelope_namespace 

106 else fallback_namespace 

107 ) 

108 return job_name, namespace 

109 

110 

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

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

113 

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

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

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

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

118 cross-references rely on set equality. 

119 """ 

120 refs: set[str] = set() 

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

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

123 if not isinstance(pod_spec, dict): 

124 return [] 

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

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

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

128 continue 

129 for entry in items: 

130 if not isinstance(entry, dict): 

131 continue 

132 image = entry.get("image") 

133 if isinstance(image, str) and image: 

134 refs.add(image) 

135 return sorted(refs) 

136 

137 

138@dataclass 

139class JobInfo: 

140 """Information about a Kubernetes job.""" 

141 

142 name: str 

143 namespace: str 

144 region: str 

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

146 created_time: datetime | None = None 

147 start_time: datetime | None = None 

148 completion_time: datetime | None = None 

149 active_pods: int = 0 

150 succeeded_pods: int = 0 

151 failed_pods: int = 0 

152 parallelism: int = 1 

153 completions: int = 1 

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

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

156 

157 @property 

158 def is_complete(self) -> bool: 

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

160 

161 @property 

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

163 if self.start_time and self.completion_time: 

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

165 if self.start_time: 

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

167 return None 

168 

169 

170class JobManager: 

171 """ 

172 Manages jobs across GCO clusters. 

173 

174 Provides: 

175 - Job submission with region targeting 

176 - Job status queries across regions 

177 - Job logs retrieval 

178 - Job deletion 

179 """ 

180 

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

182 self.config = config or get_config() 

183 self._aws_client = get_aws_client(self.config) 

184 

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

186 """ 

187 Load Kubernetes manifests from a file or directory. 

188 

189 Args: 

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

191 

192 Returns: 

193 List of manifest dictionaries 

194 """ 

195 manifests = [] 

196 path_obj = Path(path) 

197 

198 if path_obj.is_file(): 

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

200 elif path_obj.is_dir(): 

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

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

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

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

205 else: 

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

207 

208 return manifests 

209 

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

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

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

213 return safe_load_all_yaml(f, allow_aliases=False) 

214 

215 def submit_job( 

216 self, 

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

218 namespace: str | None = None, 

219 target_region: str | None = None, 

220 dry_run: bool = False, 

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

222 ) -> dict[str, Any]: 

223 """ 

224 Submit a job to GCO. 

225 

226 Args: 

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

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

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

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

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

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

233 validation enforces the allowlist. 

234 target_region: Force job to specific region 

235 dry_run: Validate without applying 

236 labels: Additional labels to add to manifests 

237 

238 Returns: 

239 Submission result dictionary 

240 """ 

241 # Load manifests if path provided 

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

243 

244 # Apply the explicit or configured namespace as a fallback only — 

245 # preserve any namespace declared by an individual manifest. 

246 effective_namespace = namespace or self.config.default_namespace 

247 for manifest in manifest_list: 

248 if "metadata" not in manifest: 

249 manifest["metadata"] = {} 

250 manifest["metadata"].setdefault("namespace", effective_namespace) 

251 

252 # Apply additional labels 

253 if labels: 

254 for manifest in manifest_list: 

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

256 manifest["metadata"] = {} 

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

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

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

260 

261 # Submit via API 

262 return self._aws_client.submit_manifests( 

263 manifests=manifest_list, 

264 namespace=effective_namespace, 

265 target_region=target_region, 

266 dry_run=dry_run, 

267 ) 

268 

269 def submit_job_direct( 

270 self, 

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

272 region: str, 

273 namespace: str | None = None, 

274 dry_run: bool = False, 

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

276 ) -> dict[str, Any]: 

277 """ 

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

279 

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

281 using kubectl. Requires: 

282 - kubectl installed and in PATH 

283 - EKS access entry configured for your IAM principal 

284 - AWS credentials with eks:DescribeCluster permission 

285 

286 Args: 

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

288 region: Target region for direct submission (required) 

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

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

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

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

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

294 dry_run: Validate without applying 

295 labels: Additional labels to add to manifests 

296 

297 Returns: 

298 Submission result dictionary 

299 """ 

300 import subprocess 

301 import tempfile 

302 import uuid 

303 

304 # Load manifests if path provided 

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

306 

307 # Apply the explicit or configured namespace as a fallback only — 

308 # preserve any namespace declared by an individual manifest. 

309 effective_namespace = namespace or self.config.default_namespace 

310 for manifest in manifest_list: 

311 if "metadata" not in manifest: 

312 manifest["metadata"] = {} 

313 manifest["metadata"].setdefault("namespace", effective_namespace) 

314 

315 # Apply additional labels 

316 if labels: 

317 for manifest in manifest_list: 

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

319 manifest["metadata"] = {} 

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

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

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

323 

324 # Get cluster name from stack 

325 stack = self._aws_client.get_regional_stack(region) 

326 if not stack: 

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

328 

329 cluster_name = stack.cluster_name 

330 

331 # Update kubeconfig for the cluster 

332 from .kubectl_helpers import update_kubeconfig 

333 

334 update_kubeconfig(cluster_name, region) 

335 

336 # Handle existing Job resources before applying 

337 warnings: list[str] = [] 

338 if not dry_run: 

339 for manifest in manifest_list: 

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

341 continue 

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

343 job_ns = manifest.get("metadata", {}).get("namespace", effective_namespace) 

344 if not job_name: 

345 continue 

346 

347 existing_status = self._get_kubectl_job_status(job_name, job_ns) 

348 if existing_status is None: 

349 # No existing job — nothing to do 

350 continue 

351 

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

353 # Finished job — safe to delete and replace 

354 subprocess.run( 

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

356 capture_output=True, 

357 text=True, 

358 ) 

359 else: 

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

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

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

363 original_name = job_name 

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

365 warnings.append( 

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

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

368 ) 

369 logger.warning( 

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

371 original_name, 

372 job_ns, 

373 new_name, 

374 ) 

375 

376 # Write manifests to temp file 

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

378 yaml.dump_all(manifest_list, f) 

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

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

381 

382 try: 

383 # Build kubectl command 

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

385 

386 if dry_run: 

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

388 

389 # Run kubectl apply 

390 result = subprocess.run( 

391 kubectl_cmd, capture_output=True, text=True 

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

393 

394 if result.returncode != 0: 

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

396 

397 # Parse output to get job name 

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

399 created_resources = [] 

400 for line in output_lines: 

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

402 created_resources.append(line) 

403 

404 # Get the actual Job identity from the submitted manifest. The 

405 # name may have been changed above to avoid an active-job collision, 

406 # and a manifest-declared namespace takes precedence over the CLI 

407 # fallback. 

408 job_name = None 

409 job_namespace = effective_namespace 

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

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

412 metadata = manifest.get("metadata", {}) 

413 job_name = metadata.get("name") 

414 job_namespace = metadata.get("namespace") or job_namespace 

415 break 

416 

417 response: dict[str, Any] = { 

418 "status": "success", 

419 "method": "kubectl", 

420 "cluster": cluster_name, 

421 "region": region, 

422 "namespace": job_namespace, 

423 "job_name": job_name, 

424 "dry_run": dry_run, 

425 "resources": created_resources, 

426 "output": result.stdout, 

427 } 

428 if warnings: 

429 response["warnings"] = warnings 

430 return response 

431 

432 finally: 

433 # Clean up temp file 

434 import os 

435 

436 os.unlink(temp_path) 

437 

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

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

440 

441 Returns: 

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

443 """ 

444 import json 

445 import subprocess 

446 

447 result = subprocess.run( 

448 [ 

449 "kubectl", 

450 "get", 

451 "job", 

452 job_name, 

453 "-n", 

454 namespace, 

455 "-o", 

456 "json", 

457 ], 

458 capture_output=True, 

459 text=True, 

460 ) 

461 if result.returncode != 0: 

462 return None # Job doesn't exist 

463 

464 try: 

465 job_data = json.loads(result.stdout) 

466 except json.JSONDecodeError, KeyError: 

467 return None 

468 

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

470 for condition in conditions: 

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

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

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

474 return "complete" 

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

476 return "failed" 

477 return "active" 

478 

479 def list_jobs( 

480 self, 

481 region: str | None = None, 

482 namespace: str | None = None, 

483 status: str | None = None, 

484 all_regions: bool = False, 

485 ) -> list[JobInfo]: 

486 """ 

487 List jobs across GCO clusters. 

488 

489 Args: 

490 region: Specific region to query 

491 namespace: Filter by namespace 

492 status: Filter by status 

493 all_regions: Query all discovered regions 

494 

495 Returns: 

496 List of JobInfo objects 

497 """ 

498 jobs = [] 

499 

500 if all_regions: 

501 # Query all discovered regional stacks 

502 stacks = self._aws_client.discover_regional_stacks() 

503 for stack_region in stacks: 

504 try: 

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

506 jobs.extend(region_jobs) 

507 except Exception as e: 

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

509 continue 

510 elif region: 

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

512 else: 

513 # Use default region 

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

515 

516 return jobs 

517 

518 def _query_jobs_in_region( 

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

520 ) -> list[JobInfo]: 

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

522 try: 

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

524 

525 jobs = [] 

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

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

528 for job_data in job_list: 

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

530 

531 return jobs 

532 except Exception as exc: 

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

534 return [] 

535 

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

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

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

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

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

541 

542 # Determine job status 

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

544 job_status = "pending" 

545 for condition in conditions: 

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

547 job_status = "succeeded" 

548 break 

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

550 job_status = "failed" 

551 break 

552 

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

554 job_status = "running" 

555 

556 # Parse timestamps 

557 created_time = None 

558 if metadata.get("creationTimestamp"): 

559 created_time = datetime.fromisoformat( 

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

561 ) 

562 

563 start_time = None 

564 if status_data.get("startTime"): 

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

566 

567 completion_time = None 

568 if status_data.get("completionTime"): 

569 completion_time = datetime.fromisoformat( 

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

571 ) 

572 

573 return JobInfo( 

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

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

576 region=region, 

577 status=job_status, 

578 created_time=created_time, 

579 start_time=start_time, 

580 completion_time=completion_time, 

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

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

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

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

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

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

587 image_refs=_extract_image_refs(spec), 

588 ) 

589 

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

591 """ 

592 Get detailed information about a specific job. 

593 

594 Args: 

595 job_name: Name of the job 

596 namespace: Namespace of the job 

597 region: Region where the job is running 

598 

599 Returns: 

600 JobInfo or None if not found 

601 """ 

602 try: 

603 response = self._aws_client.get_job_details( 

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

605 ) 

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

607 except Exception as e: 

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

609 return None 

610 

611 def get_job_logs( 

612 self, 

613 job_name: str, 

614 namespace: str, 

615 region: str | None = None, 

616 tail_lines: int = 100, 

617 follow: bool = False, 

618 since_hours: int = 24, 

619 ) -> str: 

620 """ 

621 Get logs from a job. 

622 

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

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

625 where Container Insights stores application logs. 

626 

627 Args: 

628 job_name: Name of the job 

629 namespace: Namespace of the job 

630 region: Region where the job is running 

631 tail_lines: Number of lines to return 

632 follow: Stream logs (not implemented yet) 

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

634 

635 Returns: 

636 Log content as string 

637 """ 

638 if follow: 

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

640 

641 target_region = region or self.config.default_region 

642 

643 try: 

644 return self._aws_client.get_job_logs( 

645 job_name=job_name, 

646 namespace=namespace, 

647 region=target_region, 

648 tail_lines=tail_lines, 

649 ) 

650 except RuntimeError as e: 

651 error_msg = str(e) 

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

653 if any( 

654 hint in error_msg.lower() 

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

656 ): 

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

658 try: 

659 return self._get_cloudwatch_logs( 

660 job_name=job_name, 

661 region=target_region, 

662 tail_lines=tail_lines, 

663 since_hours=since_hours, 

664 ) 

665 except Exception as cw_err: 

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

667 raise RuntimeError( 

668 f"{error_msg}\n\n" 

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

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

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

672 ) from e 

673 raise 

674 

675 def _get_cloudwatch_logs( 

676 self, 

677 job_name: str, 

678 region: str, 

679 tail_lines: int = 100, 

680 since_hours: int = 24, 

681 ) -> str: 

682 """ 

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

684 

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

686 /aws/containerinsights/{cluster_name}/application 

687 

688 Args: 

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

690 region: AWS region 

691 tail_lines: Number of log lines to return 

692 since_hours: Hours to look back (default 24) 

693 

694 Returns: 

695 Log content as string 

696 """ 

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

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

699 

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

701 

702 import time 

703 

704 now = int(time.time()) 

705 start_time = now - (since_hours * 3600) 

706 

707 query = ( 

708 f"fields @timestamp, @message " 

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

710 f"| sort @timestamp asc " 

711 f"| limit {tail_lines}" 

712 ) 

713 

714 start_query = logs_client.start_query( 

715 logGroupName=log_group, 

716 startTime=start_time, 

717 endTime=now, 

718 queryString=query, 

719 ) 

720 query_id = start_query["queryId"] 

721 

722 # Poll for results (CloudWatch Insights is async) 

723 result = None 

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

725 time.sleep(1) 

726 result = logs_client.get_query_results(queryId=query_id) 

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

728 break 

729 

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

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

732 raise RuntimeError( 

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

734 ) 

735 

736 if not result["results"]: 

737 raise RuntimeError( 

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

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

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

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

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

743 ) 

744 

745 # Extract log messages from results. 

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

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

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

749 # raw message if it's not JSON. 

750 import json as _json 

751 

752 lines = [] 

753 for row in result["results"]: 

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

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

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

757 try: 

758 parsed = _json.loads(raw) 

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

760 except ValueError, TypeError: 

761 lines.append(raw) 

762 break 

763 

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

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

766 

767 def delete_job( 

768 self, 

769 job_name: str, 

770 namespace: str, 

771 region: str | None = None, 

772 expected_uid: str | None = None, 

773 ) -> dict[str, Any]: 

774 """ 

775 Delete a job. 

776 

777 Args: 

778 job_name: Name of the job 

779 namespace: Namespace of the job 

780 region: Region where the job is running 

781 

782 Returns: 

783 Deletion result 

784 """ 

785 return self._aws_client.delete_job( 

786 job_name=job_name, 

787 namespace=namespace, 

788 region=region or self.config.default_region, 

789 expected_uid=expected_uid, 

790 ) 

791 

792 def wait_for_job( 

793 self, 

794 job_name: str, 

795 namespace: str, 

796 region: str | None = None, 

797 timeout_seconds: int = 3600, 

798 poll_interval: int = 10, 

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

800 ) -> JobInfo: 

801 """ 

802 Wait for a job to complete with progress reporting. 

803 

804 Args: 

805 job_name: Name of the job 

806 namespace: Namespace of the job 

807 region: Region where the job is running 

808 timeout_seconds: Maximum time to wait 

809 poll_interval: Seconds between status checks 

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

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

812 

813 Returns: 

814 Final JobInfo 

815 

816 Raises: 

817 TimeoutError: If job doesn't complete within timeout 

818 """ 

819 import sys 

820 import time 

821 

822 start_time = time.time() 

823 

824 while True: 

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

826 

827 if job is None: 

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

829 

830 elapsed = time.time() - start_time 

831 elapsed_str = _format_duration(int(elapsed)) 

832 

833 if job.is_complete: 

834 # Clear the progress line and return 

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

836 sys.stderr.flush() 

837 return job 

838 

839 # Build progress message 

840 pods_info = ( 

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

842 ) 

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

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

845 

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

847 

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

849 progress_callback(job, int(elapsed)) 

850 else: 

851 # Overwrite the same line on stderr 

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

853 sys.stderr.flush() 

854 

855 if elapsed >= timeout_seconds: