Coverage for cli / jobs.py: 100.00%
537 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Job management for GCO CLI.
4Provides functionality to submit, query, and manage jobs across GCO clusters.
5"""
7from collections.abc import Callable, Mapping
8from dataclasses import dataclass, field
9from datetime import UTC, datetime
10from pathlib import Path
11from typing import Any
12from urllib.parse import quote
14import requests
15import yaml
17from gco.services.manifest_processor import TRAINJOB_API_VERSION, safe_load_all_yaml
19from .aws_client import get_aws_client
20from .config import GCOConfig, get_config
22# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
23# Generated at (UTC): 2026-09-03T18:56:22Z
24# Generated from Git commit: 37fd4384775eeebf18fea3e5e085cef9645077be
25# Flowchart(s) generated from this file:
26# * ``JobManager.submit_job`` -> ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job.html``
27# (PNG: ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job.png``)
28# * ``JobManager.submit_job_sqs`` -> ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job_sqs.html``
29# (PNG: ``diagrams/code_diagrams/cli/jobs.JobManager_submit_job_sqs.png``)
30# * ``JobManager.get_job_logs`` -> ``diagrams/code_diagrams/cli/jobs.JobManager_get_job_logs.html``
31# (PNG: ``diagrams/code_diagrams/cli/jobs.JobManager_get_job_logs.png``)
32# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
33# <pyflowchart-code-diagram> END
36logger = __import__("logging").getLogger(__name__)
39def _format_duration(seconds: int) -> str:
40 """Format seconds into a human-readable duration string."""
41 if seconds < 60:
42 return f"{seconds}s"
43 minutes, secs = divmod(seconds, 60)
44 if minutes < 60:
45 return f"{minutes}m{secs:02d}s"
46 hours, mins = divmod(minutes, 60)
47 return f"{hours}h{mins:02d}m{secs:02d}s"
50def _first_manifest_namespace(manifests: list[dict[str, Any]]) -> str | None:
51 """Return the first explicit ``metadata.namespace`` found in a manifest list.
53 Used by the SQS submission path to populate the envelope ``namespace``
54 field (informational — the queue processor reads each manifest's own
55 namespace for validation). Returns None if no manifest declares one.
56 """
57 for manifest in manifests:
58 ns = manifest.get("metadata", {}).get("namespace") if isinstance(manifest, dict) else None
59 if ns:
60 return str(ns)
61 return None
64def resolve_submission_identity(
65 result: Any,
66 *,
67 fallback_name: str | None = None,
68 fallback_namespace: str | None = None,
69) -> tuple[str | None, str | None]:
70 """Resolve the submitted Job name and namespace from supported responses.
72 API submissions return resource-status dictionaries, while direct kubectl
73 submissions return a top-level ``job_name`` and a ``resources`` list of
74 human-readable strings. Only mapping-shaped resources are inspected, so
75 direct response lines can never be mistaken for response envelopes.
76 """
77 if not isinstance(result, Mapping):
78 return fallback_name, fallback_namespace
80 raw_resources = result.get("resources") or []
81 if isinstance(raw_resources, Mapping):
82 raw_resources = [raw_resources]
83 resources = [resource for resource in raw_resources if isinstance(resource, Mapping)]
84 job_resources = [
85 resource for resource in resources if str(resource.get("kind", "")).lower() == "job"
86 ]
88 explicit_job_name = result.get("job_name")
89 resource_with_name = next(
90 (resource for resource in job_resources if resource.get("name")), None
91 )
92 job_name = (
93 str(explicit_job_name)
94 if explicit_job_name
95 else str(resource_with_name.get("name"))
96 if resource_with_name is not None
97 else fallback_name
98 )
100 matching_resource = next(
101 (resource for resource in job_resources if resource.get("name") == job_name),
102 resource_with_name,
103 )
104 resource_namespace = matching_resource.get("namespace") if matching_resource else None
105 envelope_namespace = result.get("namespace")
106 namespace = (
107 str(resource_namespace)
108 if resource_namespace
109 else str(envelope_namespace)
110 if envelope_namespace
111 else fallback_namespace
112 )
113 return job_name, namespace
116def _extract_image_refs(spec: dict[str, Any]) -> list[str]:
117 """Extract container image refs from a parsed Job spec.
119 The API surface for a Job carries ``spec.template.spec.containers`` and
120 ``spec.template.spec.initContainers`` lists, each entry of which has
121 a ``name`` and an ``image`` URI. Returns an alphabetically-sorted,
122 deduplicated list so the output is stable across calls — orphan-image
123 cross-references rely on set equality.
124 """
125 refs: set[str] = set()
126 template = spec.get("template") if isinstance(spec, dict) else None
127 pod_spec = template.get("spec") if isinstance(template, dict) else None
128 if not isinstance(pod_spec, dict):
129 return []
130 for key in ("containers", "initContainers"):
131 items = pod_spec.get(key, [])
132 if not isinstance(items, list):
133 continue
134 for entry in items:
135 if not isinstance(entry, dict):
136 continue
137 image = entry.get("image")
138 if isinstance(image, str) and image:
139 refs.add(image)
140 return sorted(refs)
143def _extract_scheduling(payload: dict[str, Any]) -> dict[str, Any]:
144 """Pull the node-placement block out of a job or pods API response.
146 Absent on responses that predate the field (an older regional bridge) and
147 on the list endpoint, which deliberately does not pay for a Node read per
148 job. Both cases yield empty placement rather than an error: an absent
149 instance type is honest, a guessed one is not.
150 """
151 scheduling = payload.get("scheduling")
152 if not isinstance(scheduling, dict):
153 return {}
155 nodes = scheduling.get("nodes")
156 labels = scheduling.get("node_labels")
157 return {
158 "node_name": scheduling.get("node_name"),
159 "node_instance_type": scheduling.get("node_instance_type"),
160 "node_capacity_type": scheduling.get("node_capacity_type"),
161 "node_labels": dict(labels) if isinstance(labels, dict) else {},
162 "nodes": list(nodes) if isinstance(nodes, list) else [],
163 }
166@dataclass
167class JobInfo:
168 """Information about a Kubernetes job."""
170 name: str
171 namespace: str
172 region: str
173 status: str # "pending", "running", "succeeded", "failed"
174 created_time: datetime | None = None
175 start_time: datetime | None = None
176 completion_time: datetime | None = None
177 active_pods: int = 0
178 succeeded_pods: int = 0
179 failed_pods: int = 0
180 parallelism: int = 1
181 completions: int = 1
182 labels: dict[str, str] = field(default_factory=dict)
183 image_refs: list[str] = field(default_factory=list)
185 # Where the job's pods actually landed. A job constrained to a *set* of
186 # interchangeable instance types is placed by Karpenter within that set,
187 # so the manifest records only what the run was authorized to use. These
188 # record what it used, which is what cost reconciliation and
189 # "did this fail on the smaller box?" both need.
190 #
191 # ``node_*`` describe the earliest-created scheduled pod; ``nodes`` lists
192 # every node involved (a retried job can move between instance types) with
193 # the pods on each. All are unset when nothing is scheduled yet, when the
194 # pods have been garbage-collected, or on the list endpoint, which does not
195 # pay for a Node read per job.
196 node_name: str | None = None
197 node_instance_type: str | None = None
198 node_capacity_type: str | None = None
199 node_labels: dict[str, str] = field(default_factory=dict)
200 nodes: list[dict[str, Any]] = field(default_factory=list)
202 @property
203 def is_complete(self) -> bool:
204 return self.status in ("succeeded", "failed")
206 @property
207 def duration_seconds(self) -> int | None:
208 if self.start_time and self.completion_time:
209 return int((self.completion_time - self.start_time).total_seconds())
210 if self.start_time:
211 return int((datetime.now(UTC) - self.start_time).total_seconds())
212 return None
215class JobManager:
216 """
217 Manages jobs across GCO clusters.
219 Provides:
220 - Job submission with region targeting
221 - Job status queries across regions
222 - Job logs retrieval
223 - Job deletion
224 """
226 def __init__(self, config: GCOConfig | None = None):
227 self.config = config or get_config()
228 self._aws_client = get_aws_client(self.config)
230 def load_manifests(self, path: str) -> list[dict[str, Any]]:
231 """
232 Load Kubernetes manifests from a file or directory.
234 Args:
235 path: Path to YAML file or directory containing YAML files
237 Returns:
238 List of manifest dictionaries
239 """
240 manifests = []
241 path_obj = Path(path)
243 if path_obj.is_file():
244 manifests.extend(self._load_yaml_file(path_obj))
245 elif path_obj.is_dir():
246 for yaml_file in sorted(path_obj.glob("*.yaml")):
247 manifests.extend(self._load_yaml_file(yaml_file))
248 for yaml_file in sorted(path_obj.glob("*.yml")):
249 manifests.extend(self._load_yaml_file(yaml_file))
250 else:
251 raise FileNotFoundError(f"Path not found: {path}")
253 return manifests
255 def _load_yaml_file(self, path: Path) -> list[dict[str, Any]]:
256 """Load manifests from a single YAML file."""
257 with open(path, encoding="utf-8") as f:
258 return safe_load_all_yaml(f, allow_aliases=False)
260 def submit_job(
261 self,
262 manifests: str | list[dict[str, Any]],
263 namespace: str | None = None,
264 target_region: str | None = None,
265 dry_run: bool = False,
266 labels: dict[str, str] | None = None,
267 ) -> dict[str, Any]:
268 """
269 Submit a job to GCO.
271 Args:
272 manifests: Path to manifest file/directory or list of manifest dicts
273 namespace: Fallback namespace for manifests that don't declare
274 their own. When set, each manifest's ``metadata.namespace`` is
275 filled in only if missing — existing values are preserved so
276 users who've declared a target namespace in the manifest can
277 rely on it reaching the server untouched. Server-side
278 validation enforces the allowlist.
279 target_region: Force job to specific region
280 dry_run: Validate without applying
281 labels: Additional labels to add to manifests
283 Returns:
284 Submission result dictionary
285 """
286 # Load manifests if path provided
287 manifest_list = self.load_manifests(manifests) if isinstance(manifests, str) else manifests
289 # Apply the explicit or configured namespace as a fallback only —
290 # preserve any namespace declared by an individual manifest.
291 effective_namespace = namespace or self.config.default_namespace
292 for manifest in manifest_list:
293 if "metadata" not in manifest:
294 manifest["metadata"] = {}
295 manifest["metadata"].setdefault("namespace", effective_namespace)
297 # Apply additional labels
298 if labels:
299 for manifest in manifest_list:
300 if "labels" not in manifest["metadata"]:
301 manifest["metadata"]["labels"] = {}
302 manifest["metadata"]["labels"].update(labels)
304 # Submit via API
305 return self._aws_client.submit_manifests(
306 manifests=manifest_list,
307 namespace=effective_namespace,
308 target_region=target_region,
309 dry_run=dry_run,
310 )
312 def submit_job_direct(
313 self,
314 manifests: str | list[dict[str, Any]],
315 region: str,
316 namespace: str | None = None,
317 dry_run: bool = False,
318 labels: dict[str, str] | None = None,
319 ) -> dict[str, Any]:
320 """
321 Submit a job directly to a regional cluster using kubectl.
323 This bypasses the API Gateway and submits directly to the EKS cluster
324 using kubectl. Requires:
325 - kubectl installed and in PATH
326 - EKS access entry configured for your IAM principal
327 - AWS credentials with eks:DescribeCluster permission
329 Args:
330 manifests: Path to manifest file/directory or list of manifest dicts
331 region: Target region for direct submission (required)
332 namespace: Fallback namespace for manifests that don't declare
333 their own. When set, each manifest's ``metadata.namespace`` is
334 filled in only if missing — existing values are preserved so
335 users who've declared a target namespace in the manifest can
336 rely on it reaching ``kubectl apply`` untouched.
337 dry_run: Validate without applying
338 labels: Additional labels to add to manifests
340 Returns:
341 Submission result dictionary
342 """
343 import subprocess
344 import tempfile
345 import uuid
347 # Load manifests if path provided
348 manifest_list = self.load_manifests(manifests) if isinstance(manifests, str) else manifests
350 # Apply the explicit or configured namespace as a fallback only —
351 # preserve any namespace declared by an individual manifest.
352 effective_namespace = namespace or self.config.default_namespace
353 for manifest in manifest_list:
354 if "metadata" not in manifest:
355 manifest["metadata"] = {}
356 manifest["metadata"].setdefault("namespace", effective_namespace)
358 # Apply additional labels
359 if labels:
360 for manifest in manifest_list:
361 if "labels" not in manifest["metadata"]:
362 manifest["metadata"]["labels"] = {}
363 manifest["metadata"]["labels"].update(labels)
365 # Get cluster name from stack
366 stack = self._aws_client.get_regional_stack(region)
367 if not stack:
368 raise ValueError(f"No GCO stack found in region {region}")
370 cluster_name = stack.cluster_name
372 # Update kubeconfig for the cluster
373 from .kubectl_helpers import update_kubeconfig
375 update_kubeconfig(cluster_name, region)
377 # Handle existing Job resources before applying
378 warnings: list[str] = []
379 if not dry_run:
380 for manifest in manifest_list:
381 if manifest.get("kind") != "Job":
382 continue
383 job_name = manifest.get("metadata", {}).get("name")
384 job_ns = manifest.get("metadata", {}).get("namespace", effective_namespace)
385 if not job_name:
386 continue
388 existing_status = self._get_kubectl_job_status(job_name, job_ns)
389 if existing_status is None:
390 # No existing job — nothing to do
391 continue
393 if existing_status in ("complete", "failed"):
394 # Finished job — safe to delete and replace
395 subprocess.run(
396 ["kubectl", "delete", "job", job_name, "-n", job_ns],
397 capture_output=True,
398 text=True,
399 )
400 else:
401 # Job is still active — auto-rename to avoid collision
402 suffix = uuid.uuid4().hex[:5]
403 new_name = f"{job_name}-{suffix}"
404 original_name = job_name
405 manifest["metadata"]["name"] = new_name
406 warnings.append(
407 f"Job '{original_name}' is still running in namespace "
408 f"'{job_ns}'. Renamed new submission to '{new_name}'."
409 )
410 logger.warning(
411 "Job %s is active in %s, renamed to %s",
412 original_name,
413 job_ns,
414 new_name,
415 )
417 # Write manifests to temp file
418 with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
419 yaml.dump_all(manifest_list, f)
420 f.flush() # Ensure content is written before using f.name
421 temp_path = f.name # nosemgrep: tempfile-without-flush
423 try:
424 # Build kubectl command
425 kubectl_cmd = ["kubectl", "apply", "-f", temp_path]
427 if dry_run:
428 kubectl_cmd.extend(["--dry-run=client"])
430 # Run kubectl apply
431 result = subprocess.run(
432 kubectl_cmd, capture_output=True, text=True
433 ) # nosemgrep: dangerous-subprocess-use-audit - kubectl_cmd is a list ["kubectl","apply","-f",temp_path]; temp_path is a secure tempfile, not user input
435 if result.returncode != 0:
436 raise RuntimeError(f"kubectl apply failed: {result.stderr}")
438 # Parse output to get job name
439 output_lines = result.stdout.strip().split("\n")
440 created_resources = []
441 for line in output_lines:
442 if line:
443 created_resources.append(line)
445 # Get the actual Job identity from the submitted manifest. The
446 # name may have been changed above to avoid an active-job collision,
447 # and a manifest-declared namespace takes precedence over the CLI
448 # fallback.
449 job_name = None
450 job_namespace = effective_namespace
451 for manifest in manifest_list:
452 if manifest.get("kind") == "Job":
453 metadata = manifest.get("metadata", {})
454 job_name = metadata.get("name")
455 job_namespace = metadata.get("namespace") or job_namespace
456 break
458 response: dict[str, Any] = {
459 "status": "success",
460 "method": "kubectl",
461 "cluster": cluster_name,
462 "region": region,
463 "namespace": job_namespace,
464 "job_name": job_name,
465 "dry_run": dry_run,
466 "resources": created_resources,
467 "output": result.stdout,
468 }
469 if warnings:
470 response["warnings"] = warnings
471 return response
473 finally:
474 # Clean up temp file
475 import os
477 os.unlink(temp_path)
479 def _get_kubectl_job_status(self, job_name: str, namespace: str) -> str | None:
480 """Check the status of an existing Job via kubectl.
482 Returns:
483 "complete", "failed", "active", or None if the job doesn't exist.
484 """
485 import json
486 import subprocess
488 result = subprocess.run(
489 [
490 "kubectl",
491 "get",
492 "job",
493 job_name,
494 "-n",
495 namespace,
496 "-o",
497 "json",
498 ],
499 capture_output=True,
500 text=True,
501 )
502 if result.returncode != 0:
503 return None # Job doesn't exist
505 try:
506 job_data = json.loads(result.stdout)
507 except json.JSONDecodeError, KeyError:
508 return None
510 conditions = job_data.get("status", {}).get("conditions") or []
511 for condition in conditions:
512 cond_type = condition.get("type", "")
513 cond_status = condition.get("status", "")
514 if cond_type == "Complete" and cond_status == "True":
515 return "complete"
516 if cond_type == "Failed" and cond_status == "True":
517 return "failed"
518 return "active"
520 def list_jobs(
521 self,
522 region: str | None = None,
523 namespace: str | None = None,
524 status: str | None = None,
525 all_regions: bool = False,
526 ) -> list[JobInfo]:
527 """
528 List jobs across GCO clusters.
530 Args:
531 region: Specific region to query
532 namespace: Filter by namespace
533 status: Filter by status
534 all_regions: Query all discovered regions
536 Returns:
537 List of JobInfo objects
538 """
539 jobs = []
541 if all_regions:
542 # Query all discovered regional stacks
543 stacks = self._aws_client.discover_regional_stacks()
544 for stack_region in stacks:
545 try:
546 region_jobs = self._query_jobs_in_region(stack_region, namespace, status)
547 jobs.extend(region_jobs)
548 except Exception as e:
549 logger.warning("Failed to query jobs in %s: %s", stack_region, e)
550 continue
551 elif region:
552 jobs = self._query_jobs_in_region(region, namespace, status)
553 else:
554 # Use default region
555 jobs = self._query_jobs_in_region(self.config.default_region, namespace, status)
557 return jobs
559 def _query_jobs_in_region(
560 self, region: str, namespace: str | None, status: str | None
561 ) -> list[JobInfo]:
562 """Query jobs in a specific region."""
563 try:
564 response = self._aws_client.get_jobs(region=region, namespace=namespace, status=status)
566 jobs = []
567 # response is a list, but we expect a dict with "jobs" key from the API
568 job_list = response.get("jobs", []) if isinstance(response, dict) else response
569 for job_data in job_list:
570 jobs.append(self._parse_job_info(job_data, region))
572 return jobs
573 except Exception as exc:
574 logger.warning("Failed to query jobs in %s: %s", region, exc)
575 return []
577 def _parse_job_info(self, job_data: dict[str, Any], region: str) -> JobInfo:
578 """Parse job data into JobInfo object."""
579 metadata = job_data.get("metadata", {})
580 status_data = job_data.get("status", {})
581 spec = job_data.get("spec", {})
583 # Determine job status
584 conditions = status_data.get("conditions", [])
585 job_status = "pending"
586 for condition in conditions:
587 if condition.get("type") == "Complete" and condition.get("status") == "True":
588 job_status = "succeeded"
589 break
590 if condition.get("type") == "Failed" and condition.get("status") == "True":
591 job_status = "failed"
592 break
594 if job_status == "pending" and status_data.get("active", 0) > 0:
595 job_status = "running"
597 # Parse timestamps
598 created_time = None
599 if metadata.get("creationTimestamp"):
600 created_time = datetime.fromisoformat(
601 metadata["creationTimestamp"].replace("Z", "+00:00")
602 )
604 start_time = None
605 if status_data.get("startTime"):
606 start_time = datetime.fromisoformat(status_data["startTime"].replace("Z", "+00:00"))
608 completion_time = None
609 if status_data.get("completionTime"):
610 completion_time = datetime.fromisoformat(
611 status_data["completionTime"].replace("Z", "+00:00")
612 )
614 return JobInfo(
615 name=metadata.get("name", ""),
616 namespace=metadata.get("namespace", "default"),
617 region=region,
618 status=job_status,
619 created_time=created_time,
620 start_time=start_time,
621 completion_time=completion_time,
622 active_pods=status_data.get("active", 0),
623 succeeded_pods=status_data.get("succeeded", 0),
624 failed_pods=status_data.get("failed", 0),
625 parallelism=spec.get("parallelism", 1),
626 completions=spec.get("completions", 1),
627 labels=metadata.get("labels", {}),
628 image_refs=_extract_image_refs(spec),
629 **_extract_scheduling(job_data),
630 )
632 def get_job(self, job_name: str, namespace: str, region: str | None = None) -> JobInfo | None:
633 """
634 Get detailed information about a specific job.
636 A name the batch Job endpoint does not know (404) is retried as a
637 Kubeflow TrainJob through the generic manifests endpoint, so
638 ``gco jobs get`` / ``gco jobs submit --wait`` work unchanged for
639 TrainJobs.
641 ``None`` means the API confirmed the job does not exist (HTTP 404 on
642 both lookups). Any other failure — an unreachable regional API bridge,
643 an auth error, a 5xx — propagates to the caller so it is never
644 misreported as "not found" (see issue #258).
646 Args:
647 job_name: Name of the job
648 namespace: Namespace of the job
649 region: Region where the job is running
651 Returns:
652 JobInfo, or None if the API confirmed the job does not exist
654 Raises:
655 RuntimeError: If the regional API bridge is unreachable or the
656 request fails for reasons other than the job being absent
657 requests.exceptions.HTTPError: For non-404 HTTP responses
658 """
659 target_region = region or self.config.default_region
660 try:
661 response = self._aws_client.get_job_details(
662 job_name=job_name, namespace=namespace, region=target_region
663 )
664 return self._parse_job_info(response, target_region)
665 except requests.exceptions.HTTPError as e:
666 if e.response is not None and e.response.status_code == 404:
667 # The batch Job endpoint confirmed the name is unknown; the
668 # job may still exist as a Kubeflow TrainJob.
669 trainjob = self._get_trainjob_info(job_name, namespace, target_region)
670 if trainjob is not None:
671 return trainjob
672 logger.debug("Job %s not found in %s/%s", job_name, target_region, namespace)
673 return None
674 # Non-404 responses (401/403/5xx) do not mean the job is absent.
675 # Surface the real failure instead of collapsing it into None,
676 # which the CLI would render as "not found".
677 raise
679 def _get_trainjob_info(self, job_name: str, namespace: str, region: str) -> JobInfo | None:
680 """Fetch a TrainJob through the generic manifests endpoint, or None."""
681 try:
682 response = self._aws_client.call_api(
683 "GET",
684 f"/api/v1/manifests/{quote(namespace, safe='')}/{quote(job_name, safe='')}",
685 region=region,
686 params={"api_version": TRAINJOB_API_VERSION, "kind": "TrainJob"},
687 )
688 except RuntimeError as e:
689 # 404 (no such TrainJob either) and transport errors both mean
690 # "not resolvable as a TrainJob"; the caller reports not-found.
691 logger.debug("TrainJob fallback lookup failed for %s: %s", job_name, e)
692 return None
693 resource = response.get("resource") or {}
694 if not resource.get("exists", False):
695 return None
696 info = self._parse_trainjob_info(resource, region)
697 self._attach_trainjob_scheduling(info, job_name, namespace, region)
698 return info
700 def _attach_trainjob_scheduling(
701 self, info: JobInfo, job_name: str, namespace: str, region: str
702 ) -> None:
703 """Fill in node placement for a TrainJob from its child Job's pods.
705 The manifests endpoint that resolves a TrainJob knows nothing about
706 pods, so placement comes from the child Job ``<name>-node-0`` whose
707 indexed pods are the node ranks (same convention the log fetch uses).
708 Best effort: a TrainJob with no pods yet, or a bridge too old to report
709 placement, simply leaves the fields unset.
710 """
711 try:
712 response = self._aws_client.get_job_pods(f"{job_name}-node-0", namespace, region)
713 except Exception as exc:
714 logger.debug("No node placement available for TrainJob %s: %s", job_name, exc)
715 return
717 scheduling = _extract_scheduling(response if isinstance(response, dict) else {})
718 for attr, value in scheduling.items():
719 setattr(info, attr, value)
721 def _parse_trainjob_info(self, resource: dict[str, Any], region: str) -> JobInfo:
722 """Map a TrainJob resource payload onto JobInfo.
724 TrainJob status carries ``conditions`` (terminal types ``Complete`` /
725 ``Failed``, same strings as batch Jobs) plus per-child-Job counts in
726 ``jobsStatus``; there is no ``startTime``/``completionTime``, so the
727 terminal condition's ``lastTransitionTime`` stands in for completion.
728 """
729 metadata = resource.get("metadata", {}) or {}
730 status_data = resource.get("status", {}) or {}
731 spec = resource.get("spec", {}) or {}
733 job_status = "pending"
734 completion_time = None
735 for condition in status_data.get("conditions", []) or []:
736 if condition.get("status") != "True":
737 continue
738 if condition.get("type") == "Complete":
739 job_status = "succeeded"
740 elif condition.get("type") == "Failed":
741 job_status = "failed"
742 else:
743 continue
744 if condition.get("lastTransitionTime"):
745 completion_time = datetime.fromisoformat(
746 condition["lastTransitionTime"].replace("Z", "+00:00")
747 )
748 break
750 active = succeeded = failed = 0
751 for child in status_data.get("jobsStatus", []) or []:
752 active += child.get("active", 0) or 0
753 succeeded += child.get("succeeded", 0) or 0
754 failed += child.get("failed", 0) or 0
755 if job_status == "pending" and active > 0:
756 job_status = "running"
758 created_time = None
759 if metadata.get("creationTimestamp"):
760 created_time = datetime.fromisoformat(
761 metadata["creationTimestamp"].replace("Z", "+00:00")
762 )
764 trainer = spec.get("trainer", {}) or {}
765 try:
766 num_nodes = max(1, int(trainer.get("numNodes") or 1))
767 except TypeError, ValueError:
768 num_nodes = 1
770 return JobInfo(
771 name=metadata.get("name", ""),
772 namespace=metadata.get("namespace", "default"),
773 region=region,
774 status=job_status,
775 created_time=created_time,
776 # TrainJob status reports no startTime; leave it unset rather
777 # than fabricating one from creation time.
778 completion_time=completion_time,
779 active_pods=active,
780 succeeded_pods=succeeded,
781 failed_pods=failed,
782 parallelism=num_nodes,
783 completions=num_nodes,
784 labels=metadata.get("labels", {}) or {},
785 image_refs=sorted({trainer["image"]} if trainer.get("image") else set()),
786 )
788 def get_job_logs(
789 self,
790 job_name: str,
791 namespace: str,
792 region: str | None = None,
793 tail_lines: int = 100,
794 follow: bool = False,
795 since_hours: int = 24,
796 node: int = 0,
797 ) -> str:
798 """
799 Get logs from a job.
801 Tries the Kubernetes API first (via the GCO API). A name the batch
802 Job endpoint does not know is retried as a Kubeflow TrainJob: the
803 torch runtime runs the whole job as one child Job named
804 ``<name>-node-0`` whose indexed pods are the node ranks, so the
805 rank-``node`` pod's logs are fetched (rank 0 by default). If the pod
806 is no longer available (completed/deleted), falls back to CloudWatch
807 Logs where Container Insights stores application logs.
809 Args:
810 job_name: Name of the job
811 namespace: Namespace of the job
812 region: Region where the job is running
813 tail_lines: Number of lines to return
814 follow: Stream logs (not implemented yet)
815 since_hours: Hours to look back in CloudWatch (default 24)
816 node: Node rank to fetch for a distributed TrainJob (default 0)
818 Returns:
819 Log content as string
820 """
821 if follow:
822 raise NotImplementedError("Log streaming not yet implemented")
824 target_region = region or self.config.default_region
826 try:
827 return self._aws_client.get_job_logs(
828 job_name=job_name,
829 namespace=namespace,
830 region=target_region,
831 tail_lines=tail_lines,
832 )
833 except RuntimeError as e:
834 error_msg = str(e)
835 lowered = error_msg.lower()
836 # An unknown batch Job name may be a TrainJob; its pods run
837 # under the JobSet child Job instead.
838 if "not found" in lowered:
839 try:
840 return self._get_trainjob_node_logs(
841 job_name=job_name,
842 namespace=namespace,
843 region=target_region,
844 node=node,
845 tail_lines=tail_lines,
846 )
847 except Exception as tj_err:
848 logger.debug("TrainJob logs fallback failed: %s", tj_err)
849 # If the pod is gone or pending, try CloudWatch
850 if any(
851 hint in lowered
852 for hint in ["not found", "pending", "no pods", "terminated", "completed"]
853 ):
854 logger.info("Pod not available, falling back to CloudWatch Logs")
855 try:
856 return self._get_cloudwatch_logs(
857 job_name=job_name,
858 region=target_region,
859 tail_lines=tail_lines,
860 since_hours=since_hours,
861 )
862 except Exception as cw_err:
863 logger.debug("CloudWatch fallback failed: %s", cw_err)
864 raise RuntimeError(
865 f"{error_msg}\n\n"
866 f"CloudWatch Logs fallback also failed: {cw_err}\n"
867 f"Tip: Container logs appear in CloudWatch within a few minutes. "
868 f"If the job just finished, try again shortly."
869 ) from e
870 raise
872 def _get_trainjob_node_logs(
873 self,
874 job_name: str,
875 namespace: str,
876 region: str,
877 node: int,
878 tail_lines: int,
879 ) -> str:
880 """Fetch logs from the rank-``node`` pod of a TrainJob.
882 The shipped torch-distributed runtime materializes a TrainJob as a
883 JobSet with a single replicated Job ``<name>-node-0`` running
884 ``numNodes`` indexed pods; the completion index is the node rank
885 (``PET_NODE_RANK``). Raises if no rank-``node`` pod exists.
886 """
887 child_job = f"{job_name}-node-0"
888 pods_response = self._aws_client.get_job_pods(child_job, namespace, region)
889 pods = pods_response.get("pods", []) or []
890 if not pods:
891 raise RuntimeError(f"no pods found for TrainJob child job '{child_job}'")
893 target_pod = None
894 for pod in pods:
895 metadata = pod.get("metadata", {}) or {}
896 labels = metadata.get("labels", {}) or {}
897 if labels.get("batch.kubernetes.io/job-completion-index") == str(node):
898 target_pod = metadata.get("name")
899 break
900 if target_pod is None:
901 # Older control planes may omit the index label; fall back to the
902 # deterministic indexed-pod name prefix '<child>-<index>-'.
903 prefix = f"{child_job}-{node}-"
904 for pod in pods:
905 pod_name = (pod.get("metadata", {}) or {}).get("name", "")
906 if pod_name.startswith(prefix):
907 target_pod = pod_name
908 break
909 if target_pod is None:
910 raise RuntimeError(
911 f"no rank-{node} pod found for TrainJob '{job_name}' "
912 f"({len(pods)} pod(s) under child job '{child_job}'); "
913 f"list them with: gco jobs pods {child_job} -n {namespace} -r {region}"
914 )
916 logger.info(
917 "Job '%s' resolved as TrainJob; fetching logs from rank-%d pod '%s'",
918 job_name,
919 node,
920 target_pod,
921 )
922 response = self._aws_client.get_pod_logs(
923 job_name=child_job,
924 pod_name=target_pod,
925 namespace=namespace,
926 region=region,
927 tail_lines=tail_lines,
928 )
929 return str(response.get("logs", ""))
931 def _get_cloudwatch_logs(
932 self,
933 job_name: str,
934 region: str,
935 tail_lines: int = 100,
936 since_hours: int = 24,
937 ) -> str:
938 """
939 Fetch job logs from CloudWatch Logs (Container Insights).
941 The CloudWatch Observability addon ships container stdout/stderr to:
942 /aws/containerinsights/{cluster_name}/application
944 Args:
945 job_name: Name of the job (used to filter log streams)
946 region: AWS region
947 tail_lines: Number of log lines to return
948 since_hours: Hours to look back (default 24)
950 Returns:
951 Log content as string
952 """
953 cluster_name = f"{self.config.project_name}-{region}"
954 log_group = f"/aws/containerinsights/{cluster_name}/application"
956 logs_client = self._aws_client._session.client("logs", region_name=region)
958 import time
960 now = int(time.time())
961 start_time = now - (since_hours * 3600)
963 query = (
964 f"fields @timestamp, @message "
965 f'| filter @logStream like "{job_name}" '
966 f"| sort @timestamp asc "
967 f"| limit {tail_lines}"
968 )
970 start_query = logs_client.start_query(
971 logGroupName=log_group,
972 startTime=start_time,
973 endTime=now,
974 queryString=query,
975 )
976 query_id = start_query["queryId"]
978 # Poll for results (CloudWatch Insights is async)
979 result = None
980 for _ in range(30): # up to 30 seconds
981 time.sleep(1)
982 result = logs_client.get_query_results(queryId=query_id)
983 if result["status"] in ("Complete", "Failed", "Cancelled"):
984 break
986 if result is None or result["status"] != "Complete":
987 status = result["status"] if result else "unknown"
988 raise RuntimeError(
989 f"CloudWatch Logs query did not complete (status: {status}). Try again in a moment."
990 )
992 if not result["results"]:
993 raise RuntimeError(
994 f"No logs found in CloudWatch for job '{job_name}' "
995 f"in the last {since_hours} hours (log group: {log_group}). "
996 f"Logs may take 1-2 minutes to appear after a pod runs. "
997 f"Use --since to search further back, or check the job name "
998 f"with: gco jobs list -r {region}"
999 )
1001 # Extract log messages from results.
1002 # CloudWatch Container Insights wraps logs in a JSON envelope:
1003 # {"time":"...","stream":"stdout","log":"actual message","kubernetes":{...}}
1004 # We parse out the "log" field for clean output, falling back to the
1005 # raw message if it's not JSON.
1006 import json as _json
1008 lines = []
1009 for row in result["results"]:
1010 for entry in row:
1011 if entry["field"] == "@message":
1012 raw = entry["value"].rstrip()
1013 try:
1014 parsed = _json.loads(raw)
1015 message = parsed.get("log", raw) if isinstance(parsed, dict) else raw
1016 lines.append(message.rstrip() if isinstance(message, str) else raw)
1017 except ValueError, TypeError:
1018 lines.append(raw)
1019 break
1021 header = f"[CloudWatch Logs — {log_group}]\n"
1022 return header + "\n".join(lines)
1024 def delete_job(
1025 self,
1026 job_name: str,
1027 namespace: str,
1028 region: str | None = None,
1029 expected_uid: str | None = None,
1030 ) -> dict[str, Any]:
1031 """
1032 Delete a job.
1034 A name the batch Job endpoint does not know (404) is retried as a
1035 Kubeflow TrainJob through the generic manifests endpoint. Deleting
1036 the TrainJob cascades to its JobSet, child Jobs, and pods via owner
1037 references. ``expected_uid`` (a batch-Job concurrency guard) is not
1038 supported on the TrainJob path.
1040 Args:
1041 job_name: Name of the job
1042 namespace: Namespace of the job
1043 region: Region where the job is running
1045 Returns:
1046 Deletion result
1047 """
1048 target_region = region or self.config.default_region
1049 try:
1050 return self._aws_client.delete_job(
1051 job_name=job_name,
1052 namespace=namespace,
1053 region=target_region,
1054 expected_uid=expected_uid,
1055 )
1056 except requests.exceptions.HTTPError as e:
1057 if e.response is not None and e.response.status_code == 404:
1058 trainjob_result = self._delete_trainjob(job_name, namespace, target_region)
1059 if trainjob_result is not None:
1060 return trainjob_result
1061 raise
1063 def _delete_trainjob(self, job_name: str, namespace: str, region: str) -> dict[str, Any] | None:
1064 """Delete a TrainJob via the manifests endpoint, or None if that fails.
1066 Returning None (rather than raising) lets the caller surface the
1067 original batch-Job 404, which is the right error for a typo'd name.
1068 """
1069 try:
1070 return self._aws_client.call_api(
1071 "DELETE",
1072 f"/api/v1/manifests/{quote(namespace, safe='')}/{quote(job_name, safe='')}",
1073 region=region,
1074 params={"api_version": TRAINJOB_API_VERSION, "kind": "TrainJob"},
1075 )
1076 except RuntimeError as e:
1077 logger.debug("TrainJob delete fallback failed for %s: %s", job_name, e)
1078 return None
1080 def wait_for_job(
1081 self,
1082 job_name: str,
1083 namespace: str,
1084 region: str | None = None,
1085 timeout_seconds: int = 3600,
1086 poll_interval: int = 10,
1087 progress_callback: Callable[[JobInfo, int], None] | None = None,
1088 ) -> JobInfo:
1089 """
1090 Wait for a job to complete with progress reporting.
1092 Args:
1093 job_name: Name of the job
1094 namespace: Namespace of the job
1095 region: Region where the job is running
1096 timeout_seconds: Maximum time to wait
1097 poll_interval: Seconds between status checks
1098 progress_callback: Optional callable(JobInfo, elapsed_seconds) for progress updates.
1099 If None, a default stderr progress line is printed.
1101 Returns:
1102 Final JobInfo
1104 Raises:
1105 TimeoutError: If job doesn't complete within timeout
1106 """
1107 import sys
1108 import time
1110 start_time = time.time()
1112 while True:
1113 job = self.get_job(job_name, namespace, region)
1115 if job is None:
1116 raise ValueError(f"Job {job_name} not found in namespace {namespace}")
1118 elapsed = time.time() - start_time
1119 elapsed_str = _format_duration(int(elapsed))
1121 if job.is_complete:
1122 # Clear the progress line and return
1123 sys.stderr.write("\r\033[K")
1124 sys.stderr.flush()
1125 return job
1127 # Build progress message
1128 pods_info = (
1129 f"{job.active_pods} active, {job.succeeded_pods}/{job.completions} succeeded"
1130 )
1131 if job.failed_pods:
1132 pods_info += f", {job.failed_pods} failed"
1134 status_line = f" ⏳ {job.status.capitalize()} — {pods_info} — {elapsed_str} elapsed"
1136 if progress_callback:
1137 progress_callback(job, int(elapsed))
1138 else:
1139 # Overwrite the same line on stderr
1140 sys.stderr.write(f"\r\033[K{status_line}")
1141 sys.stderr.flush()
1143 if elapsed >= timeout_seconds:
1144 sys.stderr.write("\r\033[K")
1145 sys.stderr.flush()
1146 raise TimeoutError(
1147 f"Job {job_name} did not complete within {timeout_seconds} seconds "
1148 f"(last status: {job.status}, pods: {pods_info})"
1149 )
1151 time.sleep(poll_interval) # nosemgrep: arbitrary-sleep - intentional polling delay
1153 def submit_job_sqs(
1154 self,
1155 manifests: str | list[dict[str, Any]],
1156 region: str,
1157 namespace: str | None = None,
1158 labels: dict[str, str] | None = None,
1159 priority: int = 0,
1160 ) -> dict[str, Any]:
1161 """
1162 Submit a job to a regional SQS queue for processing.
1164 This is the recommended way to submit jobs as it:
1165 - Decouples submission from processing
1166 - Enables KEDA-based autoscaling
1167 - Provides better fault tolerance
1169 Args:
1170 manifests: Path to manifest file/directory or list of manifest dicts
1171 region: Target region for job submission (required)
1172 namespace: Fallback namespace for manifests that don't declare
1173 their own. When set, each manifest's ``metadata.namespace`` is
1174 filled in only if missing — existing values are preserved so
1175 users who've declared a target namespace in the manifest can
1176 rely on it reaching the queue processor untouched. Server-side
1177 validation enforces the allowlist.
1178 labels: Additional labels to add to manifests
1179 priority: Job priority (higher = more important)
1181 Returns:
1182 Submission result dictionary with message_id and queue info
1183 """
1184 import json
1185 import uuid
1187 import boto3
1189 # Load manifests if path provided
1190 manifest_list = self.load_manifests(manifests) if isinstance(manifests, str) else manifests
1192 # Apply namespace as a fallback only — preserve any namespace the
1193 # manifest declared itself.
1194 if namespace:
1195 for manifest in manifest_list:
1196 if "metadata" not in manifest:
1197 manifest["metadata"] = {}
1198 manifest["metadata"].setdefault("namespace", namespace)
1200 # Apply additional labels
1201 if labels:
1202 for manifest in manifest_list:
1203 if "metadata" not in manifest:
1204 manifest["metadata"] = {}
1205 if "labels" not in manifest["metadata"]:
1206 manifest["metadata"]["labels"] = {}
1207 manifest["metadata"]["labels"].update(labels)
1209 # Get queue URL from stack
1210 stack = self._aws_client.get_regional_stack(region)
1211 if not stack:
1212 raise ValueError(f"No GCO stack found in region {region}")
1214 # Get queue URL from CloudFormation outputs
1215 cfn = boto3.client("cloudformation", region_name=region)
1216 response = cfn.describe_stacks(StackName=stack.stack_name)
1217 outputs = {
1218 o["OutputKey"]: o["OutputValue"] for o in response["Stacks"][0].get("Outputs", [])
1219 }
1220 queue_url = outputs.get("JobQueueUrl")
1222 if not queue_url:
1223 raise ValueError(f"Job queue not found in stack {stack.stack_name}")
1225 # Create SQS message. The ``namespace`` field in the envelope is
1226 # informational only — the queue processor reads each manifest's
1227 # own ``metadata.namespace`` for validation and application. Report
1228 # the first manifest's namespace here so the submission response
1229 # matches reality when the user doesn't pass ``--namespace``.
1230 job_id = str(uuid.uuid4())[:8]
1231 envelope_namespace = namespace or _first_manifest_namespace(manifest_list) or "gco-jobs"
1232 message_body = {
1233 "job_id": job_id,
1234 "manifests": manifest_list,
1235 "namespace": envelope_namespace,
1236 "priority": priority,
1237 "submitted_at": datetime.now(UTC).isoformat(),
1238 }
1240 # Send to SQS
1241 sqs = boto3.client("sqs", region_name=region)
1242 response = sqs.send_message(
1243 QueueUrl=queue_url,
1244 MessageBody=json.dumps(message_body),
1245 MessageAttributes={
1246 "Priority": {"DataType": "Number", "StringValue": str(priority)},
1247 "JobId": {"DataType": "String", "StringValue": job_id},
1248 },
1249 )
1251 # Get job name from first manifest
1252 job_name = None
1253 for manifest in manifest_list:
1254 if manifest.get("kind") == "Job":
1255 job_name = manifest.get("metadata", {}).get("name")
1256 break
1258 return {
1259 "status": "queued",
1260 "method": "sqs",
1261 "message_id": response["MessageId"],
1262 "job_id": job_id,
1263 "job_name": job_name,
1264 "queue_url": queue_url,
1265 "region": region,
1266 "namespace": envelope_namespace,
1267 "priority": priority,
1268 }
1270 def get_queue_status(self, region: str) -> dict[str, Any]:
1271 """
1272 Get the status of the job queue in a region.
1274 Args:
1275 region: AWS region
1277 Returns:
1278 Queue status including message counts
1279 """
1280 import boto3
1282 stack = self._aws_client.get_regional_stack(region)
1283 if not stack:
1284 raise ValueError(f"No GCO stack found in region {region}")
1286 # Get queue URLs from CloudFormation outputs
1287 cfn = boto3.client("cloudformation", region_name=region)
1288 response = cfn.describe_stacks(StackName=stack.stack_name)
1289 outputs = {
1290 o["OutputKey"]: o["OutputValue"] for o in response["Stacks"][0].get("Outputs", [])
1291 }
1293 queue_url = outputs.get("JobQueueUrl")
1294 dlq_url = outputs.get("JobDlqUrl")
1296 if not queue_url:
1297 raise ValueError(f"Job queue not found in stack {stack.stack_name}")
1299 sqs = boto3.client("sqs", region_name=region)
1301 # Get main queue attributes
1302 queue_attrs = sqs.get_queue_attributes(
1303 QueueUrl=queue_url,
1304 AttributeNames=[
1305 "ApproximateNumberOfMessages",
1306 "ApproximateNumberOfMessagesNotVisible",
1307 "ApproximateNumberOfMessagesDelayed",
1308 ],
1309 )["Attributes"]
1311 result = {
1312 "region": region,
1313 "queue_url": queue_url,
1314 "messages_available": int(queue_attrs.get("ApproximateNumberOfMessages", 0)),
1315 "messages_in_flight": int(queue_attrs.get("ApproximateNumberOfMessagesNotVisible", 0)),
1316 "messages_delayed": int(queue_attrs.get("ApproximateNumberOfMessagesDelayed", 0)),
1317 }
1319 # Get DLQ attributes if available
1320 if dlq_url:
1321 dlq_attrs = sqs.get_queue_attributes(
1322 QueueUrl=dlq_url,
1323 AttributeNames=["ApproximateNumberOfMessages"],
1324 )["Attributes"]
1325 result["dlq_url"] = dlq_url
1326 result["dlq_messages"] = int(dlq_attrs.get("ApproximateNumberOfMessages", 0))
1328 return result
1330 def list_jobs_global(
1331 self,
1332 namespace: str | None = None,
1333 status: str | None = None,
1334 limit: int = 50,
1335 ) -> dict[str, Any]:
1336 """
1337 List jobs across all regions via the global API endpoint.
1339 This uses the cross-region aggregator Lambda to query all regional
1340 clusters in parallel and return a unified view.
1342 Args:
1343 namespace: Filter by namespace
1344 status: Filter by status
1345 limit: Maximum jobs to return
1347 Returns:
1348 Aggregated job list with region information
1349 """
1350 return self._aws_client.get_global_jobs(
1351 namespace=namespace,
1352 status=status,
1353 limit=limit,
1354 )
1356 def get_global_health(self) -> dict[str, Any]:
1357 """
1358 Get health status across all regions.
1360 Returns:
1361 Aggregated health status from all regional clusters
1362 """
1363 return self._aws_client.get_global_health()
1365 def get_global_status(self) -> dict[str, Any]:
1366 """
1367 Get cluster status across all regions.
1369 Returns:
1370 Aggregated status from all regional clusters
1371 """
1372 return self._aws_client.get_global_status()
1374 def bulk_delete_global(
1375 self,
1376 namespace: str | None = None,
1377 status: str | None = None,
1378 older_than_days: int | None = None,
1379 label_selector: str | None = None,
1380 dry_run: bool = True,
1381 ) -> dict[str, Any]:
1382 """
1383 Bulk delete jobs across all regions.
1385 Args:
1386 namespace: Filter by namespace
1387 status: Filter by status
1388 older_than_days: Delete jobs older than N days
1389 label_selector: Kubernetes label selector
1390 dry_run: If True, only return what would be deleted
1392 Returns:
1393 Deletion results from all regions
1394 """
1395 return self._aws_client.bulk_delete_global(
1396 namespace=namespace,
1397 status=status,
1398 older_than_days=older_than_days,
1399 label_selector=label_selector,
1400 dry_run=dry_run,
1401 )
1403 def get_job_events(
1404 self,
1405 job_name: str,
1406 namespace: str,
1407 region: str | None = None,
1408 ) -> dict[str, Any]:
1409 """
1410 Get Kubernetes events for a job.
1412 Args:
1413 job_name: Name of the job
1414 namespace: Namespace of the job
1415 region: Region where the job is running
1417 Returns:
1418 Events related to the job
1419 """
1420 return self._aws_client.get_job_events(
1421 job_name=job_name,
1422 namespace=namespace,
1423 region=region or self.config.default_region,
1424 )
1426 def get_job_pods(
1427 self,
1428 job_name: str,
1429 namespace: str,
1430 region: str | None = None,
1431 ) -> dict[str, Any]:
1432 """
1433 Get pods for a job.
1435 Args:
1436 job_name: Name of the job
1437 namespace: Namespace of the job
1438 region: Region where the job is running
1440 Returns:
1441 Pod details for the job
1442 """
1443 return self._aws_client.get_job_pods(
1444 job_name=job_name,
1445 namespace=namespace,
1446 region=region or self.config.default_region,
1447 )
1449 def get_pod_logs(
1450 self,
1451 job_name: str,
1452 pod_name: str,
1453 namespace: str,
1454 region: str | None = None,
1455 tail_lines: int = 100,
1456 container: str | None = None,
1457 ) -> dict[str, Any]:
1458 """
1459 Get logs from a specific pod of a job.
1461 Args:
1462 job_name: Name of the job
1463 pod_name: Name of the pod
1464 namespace: Namespace of the job
1465 region: Region where the job is running
1466 tail_lines: Number of lines to return
1467 container: Container name (for multi-container pods)
1469 Returns:
1470 Pod logs response
1471 """
1472 return self._aws_client.get_pod_logs(
1473 job_name=job_name,
1474 pod_name=pod_name,
1475 namespace=namespace,
1476 region=region or self.config.default_region,
1477 tail_lines=tail_lines,
1478 container=container,
1479 )
1481 def get_job_metrics(
1482 self,
1483 job_name: str,
1484 namespace: str,
1485 region: str | None = None,
1486 ) -> dict[str, Any]:
1487 """
1488 Get resource metrics for a job.
1490 Args:
1491 job_name: Name of the job
1492 namespace: Namespace of the job
1493 region: Region where the job is running
1495 Returns:
1496 Resource usage metrics for the job's pods
1497 """
1498 return self._aws_client.get_job_metrics(
1499 job_name=job_name,
1500 namespace=namespace,
1501 region=region or self.config.default_region,
1502 )
1504 def retry_job(
1505 self,
1506 job_name: str,
1507 namespace: str,
1508 region: str | None = None,
1509 ) -> dict[str, Any]:
1510 """
1511 Retry a failed job.
1513 Creates a new job from the failed job's spec with a new name.
1515 Args:
1516 job_name: Name of the failed job
1517 namespace: Namespace of the job
1518 region: Region where the job is running
1520 Returns:
1521 Result with new job name
1522 """
1523 return self._aws_client.retry_job(
1524 job_name=job_name,
1525 namespace=namespace,
1526 region=region or self.config.default_region,
1527 )
1529 def bulk_delete_jobs(
1530 self,
1531 namespace: str | None = None,
1532 status: str | None = None,
1533 older_than_days: int | None = None,
1534 label_selector: str | None = None,
1535 region: str | None = None,
1536 dry_run: bool = True,
1537 ) -> dict[str, Any]:
1538 """
1539 Bulk delete jobs in a region.
1541 Args:
1542 namespace: Filter by namespace
1543 status: Filter by status
1544 older_than_days: Delete jobs older than N days
1545 label_selector: Kubernetes label selector
1546 region: Target region
1547 dry_run: If True, only return what would be deleted
1549 Returns:
1550 Deletion results
1551 """
1552 return self._aws_client.bulk_delete_jobs(
1553 namespace=namespace,
1554 status=status,
1555 older_than_days=older_than_days,
1556 label_selector=label_selector,
1557 region=region or self.config.default_region,
1558 dry_run=dry_run,
1559 )
1562def get_job_manager(config: GCOConfig | None = None) -> JobManager:
1563 """Get a configured job manager instance."""
1564 return JobManager(config)