Coverage for cli / commands / jobs_cmd.py: 100.00%
748 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Job management commands."""
3import logging
4import sys
5from collections.abc import Mapping
6from typing import Any
8import click
10from ..config import GCOConfig
11from ..jobs import JobInfo, get_job_manager, resolve_submission_identity
12from ..output import confirm, format_job_table, get_output_formatter
14logger = logging.getLogger(__name__)
16pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
19def _resolve_result_namespace(result: Any, fallback: str) -> str:
20 """Pick the submitted Job namespace without assuming mapping resources."""
21 _job_name, namespace = resolve_submission_identity(result, fallback_namespace=fallback)
22 return namespace or fallback
25def _resolve_result_job_name(result: Any) -> str | None:
26 """Pick the generated/submitted Job name from a submission response."""
27 job_name, _namespace = resolve_submission_identity(result)
28 return job_name
31@click.group()
32@pass_config
33def jobs(config: Any) -> None:
34 """Manage jobs across GCO clusters."""
37@jobs.command("submit")
38@click.argument("manifest_path", type=click.Path(exists=True))
39@click.option(
40 "--namespace",
41 "-n",
42 help="Fallback namespace for manifests that don't declare their own",
43)
44@click.option("--region", "-r", "target_region", help="Target specific region")
45@click.option("--dry-run", is_flag=True, help="Validate without applying")
46@click.option(
47 "--check-policy",
48 is_flag=True,
49 help=(
50 "Before submitting, check the manifests against the policy the target "
51 "region actually enforces and report anything that would be rejected. "
52 "Advisory: findings are printed and submission continues"
53 ),
54)
55@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
56@click.option("--wait", "-w", is_flag=True, help="Wait for job completion")
57@click.option("--timeout", default=3600, help="Wait timeout in seconds")
58@pass_config
59def submit_job(
60 config: Any,
61 manifest_path: Any,
62 namespace: Any,
63 target_region: Any,
64 dry_run: Any,
65 check_policy: Any,
66 label: Any,
67 wait: Any,
68 timeout: Any,
69) -> None:
70 """Submit a job to GCO.
72 MANIFEST_PATH can be a YAML file or directory containing YAML files.
73 """
74 formatter = get_output_formatter(config)
75 job_manager = get_job_manager(config)
77 # Parse and validate labels at the CLI boundary. Silently dropping a
78 # malformed value would submit a differently labelled workload than the
79 # operator requested; values may still contain additional ``=`` bytes.
80 labels = {}
81 for lbl in label:
82 key, separator, value = lbl.partition("=")
83 if not separator or not key:
84 raise click.BadParameter(
85 "labels must use key=value with a non-empty key",
86 param_hint="--label",
87 )
88 labels[key] = value
90 if check_policy:
91 _run_pre_submit_policy_check(
92 config,
93 job_manager,
94 formatter,
95 manifest_path=manifest_path,
96 namespace=namespace,
97 target_region=target_region,
98 )
100 try:
101 result = job_manager.submit_job(
102 manifests=manifest_path,
103 namespace=namespace,
104 target_region=target_region,
105 dry_run=dry_run,
106 labels=labels if labels else None,
107 )
109 if dry_run:
110 formatter.print_success("Dry run successful - manifests are valid")
111 else:
112 formatter.print_success("Job submitted successfully")
114 # Surface any rename warnings from mapping-shaped API resources.
115 # Direct kubectl responses contain strings in ``resources``.
116 resources = result.get("resources", []) if isinstance(result, Mapping) else []
117 for resource in resources:
118 if not isinstance(resource, Mapping):
119 continue
120 msg = str(resource.get("message", ""))
121 if "renamed" in msg.lower() or "still running" in msg.lower():
122 formatter.print_warning(msg)
124 formatter.print(result)
126 # Wait for completion if requested
127 if wait and not dry_run:
128 job_name = _resolve_result_job_name(result)
129 if job_name:
130 # The API response tells us exactly where the resource landed
131 # (may differ from --namespace since the manifest's own value
132 # takes precedence). Fall back to the CLI flag or the config
133 # default only if the response didn't include a namespace.
134 resolved_ns = _resolve_result_namespace(
135 result, fallback=namespace or config.default_namespace
136 )
137 formatter.print_info(f"Waiting for job {job_name} to complete...")
138 final_job = job_manager.wait_for_job(
139 job_name=job_name,
140 namespace=resolved_ns,
141 region=target_region,
142 timeout_seconds=timeout,
143 )
144 formatter.print_success(f"Job completed with status: {final_job.status}")
146 except Exception as e:
147 formatter.print_error(f"Failed to submit job: {e}")
148 sys.exit(1)
151@jobs.command("submit-direct")
152@click.argument("manifest_path", type=click.Path(exists=True))
153@click.option("--region", "-r", required=True, help="Target region for direct submission")
154@click.option(
155 "--namespace",
156 "-n",
157 help="Fallback namespace for manifests that don't declare their own",
158)
159@click.option("--dry-run", is_flag=True, help="Validate without applying")
160@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
161@click.option("--wait", "-w", is_flag=True, help="Wait for job completion")
162@click.option("--timeout", default=3600, help="Wait timeout in seconds")
163@pass_config
164def submit_job_direct(
165 config: Any,
166 manifest_path: Any,
167 region: Any,
168 namespace: Any,
169 dry_run: Any,
170 label: Any,
171 wait: Any,
172 timeout: Any,
173) -> None:
174 """Submit a job directly to a regional cluster using kubectl.
176 This bypasses the API Gateway and submits directly to the EKS cluster.
178 REQUIREMENTS:
179 - kubectl installed and in PATH
180 - EKS access entry configured for your IAM principal
181 - AWS credentials with eks:DescribeCluster permission
183 To configure EKS access, run:
185 aws eks create-access-entry --cluster-name gco-REGION --principal-arn YOUR_ARN
187 aws eks associate-access-policy --cluster-name gco-REGION \\
188 --principal-arn YOUR_ARN \\
189 --policy-arn arn:<partition>:eks::aws:cluster-access-policy/AmazonEKSClusterAdminPolicy \\
190 --access-scope type=cluster
192 Examples:
193 gco jobs submit-direct job.yaml --region us-east-1
194 gco jobs submit-direct job.yaml -r us-west-2 -n gco-jobs --wait
195 """
196 formatter = get_output_formatter(config)
197 job_manager = get_job_manager(config)
199 # Parse and validate labels at the CLI boundary. Silently dropping a
200 # malformed value would submit a differently labelled workload than the
201 # operator requested; values may still contain additional ``=`` bytes.
202 labels = {}
203 for lbl in label:
204 key, separator, value = lbl.partition("=")
205 if not separator or not key:
206 raise click.BadParameter(
207 "labels must use key=value with a non-empty key",
208 param_hint="--label",
209 )
210 labels[key] = value
212 try:
213 formatter.print_info(f"Submitting directly to cluster in {region} via kubectl...")
215 result = job_manager.submit_job_direct(
216 manifests=manifest_path,
217 region=region,
218 namespace=namespace,
219 dry_run=dry_run,
220 labels=labels if labels else None,
221 )
223 if dry_run:
224 formatter.print_success("Dry run successful - manifests are valid")
225 else:
226 formatter.print_success(f"Job submitted directly to {region}")
228 # Surface any warnings (e.g. job was renamed due to name collision)
229 # without mutating or assuming the shape of the direct result.
230 warnings = result.get("warnings", []) if isinstance(result, Mapping) else []
231 for warning in warnings:
232 formatter.print_warning(str(warning))
234 formatter.print(result)
236 # Wait for completion if requested
237 if wait and not dry_run:
238 job_name = _resolve_result_job_name(result)
239 if job_name:
240 resolved_ns = _resolve_result_namespace(
241 result, fallback=namespace or config.default_namespace
242 )
243 formatter.print_info(f"Waiting for job {job_name} to complete...")
244 final_job = job_manager.wait_for_job(
245 job_name=job_name,
246 namespace=resolved_ns,
247 region=region,
248 timeout_seconds=timeout,
249 )
250 formatter.print_success(f"Job completed with status: {final_job.status}")
252 except Exception as e:
253 formatter.print_error(f"Failed to submit job directly: {e}")
254 sys.exit(1)
257@jobs.command("submit-sqs")
258@click.argument("manifest_path", type=click.Path(exists=True))
259@click.option("--region", "-r", help="Target region (auto-selects optimal if not specified)")
260@click.option(
261 "--namespace",
262 "-n",
263 help="Fallback namespace for manifests that don't declare their own",
264)
265@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
266@click.option("--priority", "-p", default=0, help="Job priority (higher = more important)")
267@click.option("--auto-region", is_flag=True, help="Auto-select optimal region based on capacity")
268@pass_config
269def submit_job_sqs(
270 config: Any,
271 manifest_path: Any,
272 region: Any,
273 namespace: Any,
274 label: Any,
275 priority: Any,
276 auto_region: Any,
277) -> None:
278 """Submit a job to a regional SQS queue for processing.
280 This is the recommended way to submit jobs as it:
281 - Decouples submission from processing
282 - Enables KEDA-based autoscaling
283 - Provides better fault tolerance
285 If --auto-region is specified, the CLI will analyze capacity across all
286 regions and submit to the optimal one.
288 Examples:
289 gco jobs submit-sqs job.yaml --region us-east-1
290 gco jobs submit-sqs job.yaml --auto-region
291 gco jobs submit-sqs job.yaml -r us-west-2 --priority 10
292 """
293 formatter = get_output_formatter(config)
294 job_manager = get_job_manager(config)
296 # Parse and validate labels at the CLI boundary. Silently dropping a
297 # malformed value would submit a differently labelled workload than the
298 # operator requested; values may still contain additional ``=`` bytes.
299 labels = {}
300 for lbl in label:
301 key, separator, value = lbl.partition("=")
302 if not separator or not key:
303 raise click.BadParameter(
304 "labels must use key=value with a non-empty key",
305 param_hint="--label",
306 )
307 labels[key] = value
309 try:
310 # Auto-select region if requested
311 if auto_region and not region:
312 formatter.print_info("Analyzing capacity across regions...")
313 from ..capacity import get_capacity_checker
315 checker = get_capacity_checker(config)
316 recommendation = checker.recommend_region_for_job()
317 region = recommendation["region"]
318 formatter.print_info(f"Selected region: {region} ({recommendation['reason']})")
319 elif not region:
320 region = config.default_region
322 formatter.print_info(f"Submitting job to SQS queue in {region}...")
324 result = job_manager.submit_job_sqs(
325 manifests=manifest_path,
326 region=region,
327 namespace=namespace,
328 labels=labels if labels else None,
329 priority=priority,
330 )
332 formatter.print_success(f"Job queued successfully in {region}")
333 formatter.print(result)
335 except Exception as e:
336 formatter.print_error(f"Failed to submit job to SQS: {e}")
337 sys.exit(1)
340@jobs.command("queue-status")
341@click.option("--region", "-r", help="Specific region to check")
342@click.option("--all-regions", "-a", is_flag=True, help="Check all regions")
343@pass_config
344def queue_status(config: Any, region: Any, all_regions: Any) -> None:
345 """Show job queue status across regions.
347 Displays the number of pending, in-flight, and failed messages
348 in the job queues.
350 Examples:
351 gco jobs queue-status --region us-east-1
352 gco jobs queue-status --all-regions
353 """
354 formatter = get_output_formatter(config)
355 job_manager = get_job_manager(config)
357 try:
358 if all_regions:
359 from ..aws_client import get_aws_client
361 aws_client = get_aws_client(config)
362 stacks = aws_client.discover_regional_stacks()
364 results = []
365 for stack_region in stacks:
366 try:
367 status = job_manager.get_queue_status(stack_region)
368 results.append(status)
369 except Exception as e:
370 logger.debug("Failed to get queue status for %s: %s", stack_region, e)
371 continue
373 if not results:
374 formatter.print_warning("No queue status available")
375 return
377 # Format as table
378 print("\n REGION PENDING IN-FLIGHT DELAYED DLQ")
379 print(" " + "-" * 55)
380 for r in results:
381 dlq = r.get("dlq_messages", 0)
382 print(
383 f" {r['region']:<15} {r['messages_available']:>7} "
384 f"{r['messages_in_flight']:>9} {r['messages_delayed']:>7} {dlq:>3}"
385 )
386 else:
387 target_region = region or config.default_region
388 status = job_manager.get_queue_status(target_region)
389 formatter.print(status)
391 except Exception as e:
392 formatter.print_error(f"Failed to get queue status: {e}")
393 sys.exit(1)
396@jobs.command("list")
397@click.option("--namespace", "-n", help="Filter by namespace")
398@click.option("--region", "-r", help="Target region (required unless --all-regions)")
399@click.option("--status", "-s", type=click.Choice(["pending", "running", "succeeded", "failed"]))
400@click.option("--all-regions", "-a", is_flag=True, help="Query all regions via global API")
401@click.option("--limit", "-l", default=50, help="Maximum jobs to return")
402@pass_config
403def list_jobs(
404 config: Any, namespace: Any, region: Any, status: Any, all_regions: Any, limit: Any
405) -> None:
406 """List jobs in GCO clusters.
408 You must specify either --region for a specific cluster or --all-regions
409 to query all clusters via the global aggregation API.
411 Examples:
412 gco jobs list --region us-east-1
413 gco jobs list --all-regions
414 gco jobs list -r us-west-2 -n gco-jobs --status running
415 """
416 formatter = get_output_formatter(config)
417 job_manager = get_job_manager(config)
419 # Require explicit region or --all-regions
420 if not region and not all_regions:
421 formatter.print_error("You must specify --region or --all-regions")
422 formatter.print_info(" Use --region/-r to query a specific cluster")
423 formatter.print_info(" Use --all-regions/-a to query all clusters")
424 sys.exit(1)
426 try:
427 if all_regions:
428 # Use global aggregation API
429 result = job_manager.list_jobs_global(
430 namespace=namespace,
431 status=status,
432 limit=limit,
433 )
435 if config.output_format == "table":
436 # Print summary
437 print("\n Global Jobs Summary")
438 print(" " + "-" * 50)
439 print(f" Total jobs: {result.get('total', 0)}")
440 print(f" Regions queried: {result.get('regions_queried', 0)}")
441 print(f" Regions successful: {result.get('regions_successful', 0)}")
443 # Print region summaries
444 if result.get("region_summaries"):
445 print("\n REGION COUNT TOTAL")
446 print(" " + "-" * 35)
447 for r in result["region_summaries"]:
448 print(f" {r['region']:<15} {r['count']:>5} {r['total']:>5}")
450 # Print jobs
451 jobs_data = result.get("jobs", [])
452 if jobs_data:
453 print(
454 "\n NAME NAMESPACE REGION STATUS"
455 )
456 print(" " + "-" * 75)
457 for job in jobs_data[:limit]:
458 name = job.get("metadata", {}).get("name", "")[:30]
459 ns = job.get("metadata", {}).get("namespace", "")[:14]
460 job_region = job.get("_source_region", "")[:14]
461 job_status = job.get("computed_status", "unknown")[:10]
462 print(f" {name:<30} {ns:<15} {job_region:<15} {job_status}")
464 # Print errors if any
465 if result.get("errors"):
466 print("\n Errors:")
467 for err in result["errors"]:
468 formatter.print_warning(f" {err['region']}: {err['error']}")
469 else:
470 formatter.print(result)
471 else:
472 # Query specific region
473 jobs_list = job_manager.list_jobs(
474 region=region, namespace=namespace, status=status, all_regions=False
475 )
477 if config.output_format == "table":
478 print(format_job_table(jobs_list))
479 else:
480 formatter.print(jobs_list)
482 except Exception as e:
483 formatter.print_error(f"Failed to list jobs: {e}")
484 sys.exit(1)
487def _print_job_placement(job: JobInfo) -> None:
488 """Print where the job's pods landed, and on what hardware.
490 The wide generic table renders ``node_labels``/``nodes`` as placeholders,
491 so the hardware — the reason those fields exist — gets its own block.
492 """
493 if not job.nodes and not job.node_name:
494 return
496 print("\n Placement")
497 print(" " + "-" * 78)
498 print(" NODE INSTANCE TYPE CAPACITY PODS")
499 print(" " + "-" * 78)
500 for node in job.nodes or [{"name": job.node_name, "pods": []}]:
501 name = str(node.get("name") or "-")[:40]
502 instance_type = str(node.get("instance_type") or "-")[:17]
503 capacity_type = str(node.get("capacity_type") or "-")[:10]
504 pod_count = len(node.get("pods") or [])
505 print(f" {name:<41} {instance_type:<17} {capacity_type:<10} {pod_count}")
507 if job.node_labels:
508 for key in sorted(job.node_labels):
509 print(f" {key}: {job.node_labels[key]}")
512def _print_unscheduled_pods(job: JobInfo) -> None:
513 """Print the pods that have no node yet, and why Kubernetes says so.
515 The Job's ``status`` says ``running`` as long as a pod exists, placed or
516 not; this block is what tells an operator the run is actually waiting on
517 capacity. Absent (not zero) counts come from a bridge that predates the
518 fields, so nothing is printed rather than a wrong verdict.
519 """
520 if not job.unscheduled_pods and not job.unscheduled:
521 return
522 print("\n Unscheduled pods")
523 print(" " + "-" * 78)
524 entries = job.unscheduled or [{}] * (job.unscheduled_pods or 0)
525 for pod in entries:
526 name = str(pod.get("name") or "-")[:60]
527 phase = str(pod.get("phase") or "-")
528 reason = str(pod.get("reason") or "-")
529 print(f" {name:<61} {phase}")
530 print(f" reason: {reason}")
531 if pod.get("since"):
532 print(f" since: {pod['since']}")
533 if pod.get("message"):
534 print(f" message: {pod['message']}")
537@jobs.command("get")
538@click.argument("job_name")
539@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
540@click.option("--region", "-r", required=True, help="Job region (required)")
541@pass_config
542def get_job(config: Any, job_name: Any, namespace: Any, region: Any) -> None:
543 """Get details of a specific job.
545 Reports the node each pod landed on along with that node's instance type
546 and spot/on-demand capacity type, so a job authorized to run on a set of
547 interchangeable instance types shows which one it actually used. Pods that
548 have no node yet are listed with the scheduler's own reason, so a Job that
549 is waiting on capacity is distinguishable from one that is training.
551 Examples:
552 gco jobs get my-job --region us-east-1
553 gco jobs get training-job -r us-west-2 -n ml-jobs
554 """
555 formatter = get_output_formatter(config)
556 job_manager = get_job_manager(config)
558 try:
559 job = job_manager.get_job(job_name, namespace, region)
560 if job:
561 formatter.print(job)
562 if config.output_format == "table" and isinstance(job, JobInfo):
563 _print_job_placement(job)
564 _print_unscheduled_pods(job)
565 else:
566 formatter.print_error(f"Job {job_name} not found")
567 sys.exit(1)
568 except Exception as e:
569 formatter.print_error(f"Failed to get job: {e}")
570 sys.exit(1)
573@jobs.command("logs")
574@click.argument("job_name")
575@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
576@click.option("--region", "-r", required=True, help="Job region (required)")
577@click.option("--tail", "-t", default=100, help="Number of lines to show")
578@click.option(
579 "--since", "-s", default=24, type=int, help="Hours to look back in CloudWatch (default: 24)"
580)
581@click.option("--container", "-c", help="Container name (for multi-container pods)")
582@click.option(
583 "--node",
584 default=0,
585 type=int,
586 help="Node rank to fetch for a distributed TrainJob (default: 0)",
587)
588@pass_config
589def get_logs(
590 config: Any,
591 job_name: Any,
592 namespace: Any,
593 region: Any,
594 tail: Any,
595 since: Any,
596 container: Any,
597 node: Any,
598) -> None:
599 """Get logs from a job.
601 Fetches logs from the Kubernetes API if the pod is still running.
602 If the pod is gone, falls back to CloudWatch Logs automatically.
603 Use --since to control how far back CloudWatch searches.
605 Kubeflow TrainJobs are resolved automatically; use --node to pick a
606 node rank other than 0.
608 Examples:
609 gco jobs logs my-job --region us-east-1
610 gco jobs logs training-job -r us-west-2 -n ml-jobs --tail 500
611 gco jobs logs old-job -r us-east-1 --since 72
612 gco jobs logs multi-container-job -r us-east-1 --container sidecar
613 gco jobs logs my-trainjob -r us-east-1 --node 1
614 """
615 formatter = get_output_formatter(config)
616 job_manager = get_job_manager(config)
618 try:
619 logs = job_manager.get_job_logs(
620 job_name, namespace, region, tail_lines=tail, since_hours=since, node=node
621 )
622 print(logs)
623 except Exception as e:
624 formatter.print_error(f"Failed to get logs: {e}")
625 sys.exit(1)
628@jobs.command("delete")
629@click.argument("job_name")
630@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
631@click.option("--region", "-r", required=True, help="Job region (required)")
632@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
633@pass_config
634def delete_job(config: Any, job_name: Any, namespace: Any, region: Any, yes: Any) -> None:
635 """Delete a job.
637 Examples:
638 gco jobs delete my-job --region us-east-1
639 gco jobs delete old-job -r us-west-2 -n ml-jobs -y
640 """
641 formatter = get_output_formatter(config)
642 job_manager = get_job_manager(config)
644 if not yes:
645 confirm(
646 f"Delete job {job_name} in namespace {namespace} ({region})?",
647 abort=True,
648 err=config.output_format != "table",
649 )
651 try:
652 result = job_manager.delete_job(job_name, namespace, region)
653 if config.output_format == "table":
654 formatter.print_success(f"Job {job_name} deleted")
655 else:
656 formatter.print(
657 {
658 **result,
659 "deleted": True,
660 "job_name": job_name,
661 "namespace": namespace,
662 "region": region,
663 }
664 )
665 except Exception as e:
666 formatter.print_error(f"Failed to delete job: {e}")
667 sys.exit(1)
670@jobs.command("events")
671@click.argument("job_name")
672@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
673@click.option("--region", "-r", required=True, help="Job region (required)")
674@pass_config
675def get_job_events(config: Any, job_name: Any, namespace: Any, region: Any) -> None:
676 """Get Kubernetes events for a job.
678 Shows events related to the job and its pods, useful for debugging
679 scheduling issues, resource problems, or startup failures.
681 Examples:
682 gco jobs events my-job --region us-east-1
683 gco jobs events training-job -n ml-jobs -r us-west-2
684 """
685 formatter = get_output_formatter(config)
686 job_manager = get_job_manager(config)
688 try:
689 result = job_manager.get_job_events(job_name, namespace, region)
691 if config.output_format == "table":
692 events = result.get("events", [])
693 if not events:
694 formatter.print_info("No events found for this job")
695 return
697 print(f"\n Events for {job_name} ({result.get('count', 0)} total)")
698 print(" " + "-" * 70)
699 for event in events:
700 event_type = event.get("type") or "Normal"
701 reason = (event.get("reason") or "")[:20]
702 message = (event.get("message") or "")[:50]
703 timestamp = (event.get("lastTimestamp") or event.get("firstTimestamp") or "")[:19]
704 marker = "⚠" if event_type == "Warning" else "✓"
705 print(f" {marker} [{timestamp}] {reason:<20} {message}")
706 else:
707 formatter.print(result)
709 except Exception as e:
710 formatter.print_error(f"Failed to get job events: {e}")
711 sys.exit(1)
714@jobs.command("pods")
715@click.argument("job_name")
716@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
717@click.option("--region", "-r", required=True, help="Job region (required)")
718@pass_config
719def get_job_pods(config: Any, job_name: Any, namespace: Any, region: Any) -> None:
720 """Get pod details for a job.
722 Shows all pods created by the job with their status, node placement,
723 the instance type each node is, and container information.
725 Examples:
726 gco jobs pods my-job -r us-east-1
727 gco jobs pods training-job -n ml-jobs -r us-west-2
728 """
729 formatter = get_output_formatter(config)
730 job_manager = get_job_manager(config)
732 try:
733 result = job_manager.get_job_pods(job_name, namespace, region)
735 if config.output_format == "table":
736 pods = result.get("pods", [])
737 if not pods:
738 formatter.print_info("No pods found for this job")
739 return
741 print(f"\n Pods for {job_name} ({result.get('count', 0)} total)")
742 print(" " + "-" * 96)
743 print(
744 " NAME NODE "
745 "INSTANCE TYPE STATUS RESTARTS"
746 )
747 print(" " + "-" * 96)
748 for pod in pods:
749 name = (pod.get("metadata", {}).get("name") or "")[:40]
750 node = (pod.get("spec", {}).get("nodeName") or "")[:22]
751 node_info = pod.get("node") or {}
752 instance_type = str(node_info.get("instance_type") or "-")[:17]
753 phase = (pod.get("status", {}).get("phase") or "Unknown")[:10]
754 restarts = sum(
755 c.get("restartCount", 0)
756 for c in (pod.get("status", {}).get("containerStatuses") or [])
757 )
758 print(f" {name:<40} {node:<23} {instance_type:<17} {phase:<10} {restarts}")
759 else:
760 formatter.print(result)
762 except Exception as e:
763 formatter.print_error(f"Failed to get job pods: {e}")
764 sys.exit(1)
767@jobs.command("pod-logs")
768@click.argument("job_name")
769@click.argument("pod_name")
770@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
771@click.option("--region", "-r", required=True, help="Job region (required)")
772@click.option("--tail", "-t", default=100, help="Number of lines to show")
773@click.option("--container", "-c", help="Container name (for multi-container pods)")
774@pass_config
775def get_pod_logs_cmd(
776 config: Any,
777 job_name: Any,
778 pod_name: Any,
779 namespace: Any,
780 region: Any,
781 tail: Any,
782 container: Any,
783) -> None:
784 """Get logs from a specific pod of a job.
786 Use 'gco jobs pods' first to list available pods, then use this
787 command to get logs from a specific pod.
789 Examples:
790 gco jobs pod-logs my-job my-job-abc123 -r us-east-1
791 gco jobs pod-logs training-job training-job-xyz789 -r us-west-2 --tail 500
792 gco jobs pod-logs multi-job multi-job-pod1 -r us-east-1 --container sidecar
793 """
794 formatter = get_output_formatter(config)
795 job_manager = get_job_manager(config)
797 try:
798 result = job_manager.get_pod_logs(
799 job_name=job_name,
800 pod_name=pod_name,
801 namespace=namespace,
802 region=region,
803 tail_lines=tail,
804 container=container,
805 )
807 # Print logs directly
808 logs = result.get("logs", "")
809 if logs:
810 print(logs)
811 else:
812 formatter.print_info("No logs available")
814 except Exception as e:
815 formatter.print_error(f"Failed to get pod logs: {e}")
816 sys.exit(1)
819@jobs.command("metrics")
820@click.argument("job_name")
821@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
822@click.option("--region", "-r", required=True, help="Job region (required)")
823@pass_config
824def get_job_metrics(config: Any, job_name: Any, namespace: Any, region: Any) -> None:
825 """Get resource usage metrics for a job.
827 Shows CPU and memory usage for all pods in the job. Requires
828 metrics-server to be installed in the cluster.
830 Examples:
831 gco jobs metrics my-job --region us-east-1
832 gco jobs metrics training-job -n ml-jobs -r us-west-2
833 """
834 formatter = get_output_formatter(config)
835 job_manager = get_job_manager(config)
837 try:
838 result = job_manager.get_job_metrics(job_name, namespace, region)
840 if config.output_format == "table":
841 summary = result.get("summary", {})
842 pods = result.get("pods", [])
844 print(f"\n Resource Metrics for {job_name}")
845 print(" " + "-" * 50)
846 print(f" Total CPU: {summary.get('total_cpu_millicores', 0)}m")
847 print(f" Total Memory: {summary.get('total_memory_mib', 0):.1f} MiB")
848 print(f" Pod Count: {summary.get('pod_count', 0)}")
850 if pods:
851 print("\n POD CPU(m) MEMORY(MiB)")
852 print(" " + "-" * 65)
853 for pod in pods:
854 pod_name = pod.get("pod_name", "")[:40]
855 cpu = sum(c.get("cpu_millicores", 0) for c in pod.get("containers", []))
856 mem = sum(c.get("memory_mib", 0) for c in pod.get("containers", []))
857 print(f" {pod_name:<40} {cpu:>6} {mem:>10.1f}")
858 else:
859 formatter.print(result)
861 except Exception as e:
862 formatter.print_error(f"Failed to get job metrics: {e}")
863 sys.exit(1)
866@jobs.command("retry")
867@click.argument("job_name")
868@click.option("--namespace", "-n", default="gco-jobs", help="Job namespace")
869@click.option("--region", "-r", required=True, help="Job region (required)")
870@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
871@pass_config
872def retry_job(config: Any, job_name: Any, namespace: Any, region: Any, yes: Any) -> None:
873 """Retry a failed job.
875 Creates a new job from the failed job's spec with a new name.
876 The original job is preserved for debugging.
878 Examples:
879 gco jobs retry failed-job --region us-east-1
880 gco jobs retry training-job -n ml-jobs -r us-west-2 -y
881 """
882 formatter = get_output_formatter(config)
883 job_manager = get_job_manager(config)
885 if not yes:
886 confirm(f"Retry job {job_name} in namespace {namespace} ({region})?", abort=True)
888 try:
889 result = job_manager.retry_job(job_name, namespace, region)
891 if result.get("success"):
892 formatter.print_success(f"Job retry created: {result.get('new_job')}")
893 else:
894 formatter.print_error(f"Failed to retry job: {result.get('message')}")
895 sys.exit(1)
897 formatter.print(result)
899 except Exception as e:
900 formatter.print_error(f"Failed to retry job: {e}")
901 sys.exit(1)
904@jobs.command("bulk-delete")
905@click.option("--namespace", "-n", help="Filter by namespace")
906@click.option("--status", "-s", type=click.Choice(["completed", "succeeded", "failed"]))
907@click.option("--older-than-days", "-d", type=int, help="Delete jobs older than N days")
908@click.option("--label-selector", "-l", help="Kubernetes label selector")
909@click.option("--region", "-r", help="Target region (required unless --all-regions)")
910@click.option("--all-regions", "-a", is_flag=True, help="Delete across all regions")
911@click.option("--dry-run", is_flag=True, default=True, help="Only show what would be deleted")
912@click.option("--execute", is_flag=True, help="Actually delete (disables dry-run)")
913@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
914@pass_config
915def bulk_delete_jobs(
916 config: Any,
917 namespace: Any,
918 status: Any,
919 older_than_days: Any,
920 label_selector: Any,
921 region: Any,
922 all_regions: Any,
923 dry_run: Any,
924 execute: Any,
925 yes: Any,
926) -> None:
927 """Bulk delete jobs based on filters.
929 You must specify either --region for a specific cluster or --all-regions
930 to delete across all clusters.
932 By default runs in dry-run mode. Use --execute to actually delete.
934 Examples:
935 gco jobs bulk-delete --region us-east-1 --status completed --older-than-days 7
936 gco jobs bulk-delete -r us-west-2 -n gco-jobs -s failed --execute -y
937 gco jobs bulk-delete --all-regions --status failed --older-than-days 30 --execute
938 """
939 formatter = get_output_formatter(config)
940 job_manager = get_job_manager(config)
942 # Require explicit region or --all-regions
943 if not region and not all_regions:
944 formatter.print_error("You must specify --region or --all-regions")
945 formatter.print_info(" Use --region/-r to delete from a specific cluster")
946 formatter.print_info(" Use --all-regions/-a to delete across all clusters")
947 sys.exit(1)
949 # --execute disables dry-run
950 if execute:
951 dry_run = False
953 if not dry_run and not yes:
954 scope = f"region {region}" if region else "ALL regions"
955 confirm(f"This will permanently delete matching jobs in {scope}. Continue?", abort=True)
957 try:
958 if region:
959 # Single region delete
960 result = job_manager.bulk_delete_jobs(
961 namespace=namespace,
962 status=status,
963 older_than_days=older_than_days,
964 label_selector=label_selector,
965 region=region,
966 dry_run=dry_run,
967 )
968 else:
969 # Global delete across all regions
970 result = job_manager.bulk_delete_global(
971 namespace=namespace,
972 status=status,
973 older_than_days=older_than_days,
974 label_selector=label_selector,
975 dry_run=dry_run,
976 )
978 if dry_run:
979 formatter.print_info("DRY RUN - No jobs were deleted")
980 formatter.print_info(f"Would delete {result.get('total_matched', 0)} jobs")
981 else:
982 formatter.print_success(
983 f"Deleted {result.get('deleted_count', result.get('total_deleted', 0))} jobs"
984 )
986 formatter.print(result)
988 except Exception as e:
989 formatter.print_error(f"Failed to bulk delete jobs: {e}")
990 sys.exit(1)
993@jobs.command("health")
994@click.option("--region", "-r", help="Target region (required unless --all-regions)")
995@click.option("--all-regions", "-a", is_flag=True, help="Get health across all regions")
996@pass_config
997def job_health(config: Any, region: Any, all_regions: Any) -> None:
998 """Get health status of GCO clusters.
1000 You must specify either --region for a specific cluster or --all-regions
1001 to get health status across all clusters.
1003 Examples:
1004 gco jobs health --region us-east-1
1005 gco jobs health --all-regions
1006 """
1007 formatter = get_output_formatter(config)
1008 job_manager = get_job_manager(config)
1010 # Require explicit region or --all-regions
1011 if not region and not all_regions:
1012 formatter.print_error("You must specify --region or --all-regions")
1013 formatter.print_info(" Use --region/-r to check a specific cluster")
1014 formatter.print_info(" Use --all-regions/-a to check all clusters")
1015 sys.exit(1)
1017 try:
1018 if all_regions:
1019 result = job_manager.get_global_health()
1021 if config.output_format == "table":
1022 print(
1023 f"\n Global Health Status: {result.get('overall_status', 'unknown').upper()}"
1024 )
1025 print(" " + "-" * 50)
1026 print(
1027 f" Healthy regions: {result.get('healthy_regions', 0)}/{result.get('total_regions', 0)}"
1028 )
1030 regions = result.get("regions", [])
1031 if regions:
1032 print("\n REGION STATUS CLUSTER")
1033 print(" " + "-" * 50)
1034 for r in regions:
1035 status_icon = "✓" if r.get("status") == "healthy" else "✗"
1036 print(
1037 f" {status_icon} {r.get('region', ''):<13} {r.get('status', ''):<12} {r.get('cluster_id', '')}"
1038 )
1039 else:
1040 formatter.print(result)
1041 else:
1042 # Single region health check via API
1043 result = job_manager._aws_client.get_health(region=region)
1044 formatter.print(result)
1046 except Exception as e:
1047 formatter.print_error(f"Failed to get health status: {e}")
1048 sys.exit(1)
1051@jobs.command("policy")
1052@click.option("--region", "-r", required=True, help="Target region")
1053@pass_config
1054def job_policy(config: Any, region: Any) -> None:
1055 """Show the job validation policy a region actually enforces.
1057 Reads the deployed manifest processor, not your local cdk.json — those
1058 diverge whenever the stack was deployed from a different checkout, and CDK
1059 adds the project's own ECR registries to the trusted list at synth time.
1061 Use this before submitting to know whether a manifest will be admitted.
1062 Three layers must all pass: the front-door policy, the namespace
1063 LimitRange (per container), and the namespace ResourceQuota (aggregate).
1065 Examples:
1066 gco jobs policy --region us-east-1
1067 gco jobs policy -r us-east-1 -o json
1068 """
1069 formatter = get_output_formatter(config)
1070 job_manager = get_job_manager(config)
1072 try:
1073 result = job_manager._aws_client.get_job_validation_policy(region=region)
1075 if config.output_format != "table":
1076 formatter.print(result)
1077 return
1079 policy = result.get("policy", {})
1080 caps = policy.get("manifest_caps", {})
1082 print(f"\n Job Validation Policy — {result.get('region', region)}")
1083 print(f" Cluster: {result.get('cluster_id', 'unknown')}")
1084 print(" " + "-" * 60)
1085 print(f" Validation enabled: {policy.get('validation_enabled')}")
1086 print("\n PER-MANIFEST CAPS (front door)")
1087 print(f" max CPU: {caps.get('max_cpu_millicores')}m")
1088 print(f" max memory: {caps.get('max_memory_bytes')} bytes")
1089 print(f" max GPU: {caps.get('max_gpu_count')}")
1090 print("\n ALLOWLISTS")
1091 print(f" namespaces: {', '.join(policy.get('allowed_namespaces', []))}")
1092 print(f" kinds: {', '.join(policy.get('allowed_kinds', []))}")
1093 print(f" registries: {', '.join(policy.get('trusted_registries', []))}")
1094 print(f" dockerhub orgs: {', '.join(policy.get('trusted_dockerhub_orgs', []))}")
1095 print("\n OTHER CHECKS")
1096 print(
1097 f" accelerator toleration required: {policy.get('require_accelerator_toleration')}"
1098 )
1099 print(f" YAML max depth: {policy.get('yaml_max_depth')}")
1101 security = policy.get("manifest_security_policy", {})
1102 if security:
1103 enabled = sorted(name for name, on in security.items() if on)
1104 disabled = sorted(name for name, on in security.items() if not on)
1105 print(f" blocked: {', '.join(enabled) if enabled else 'none'}")
1106 print(f" not blocked: {', '.join(disabled) if disabled else 'none'}")
1108 enforcement = result.get("cluster_enforcement", {})
1109 if enforcement:
1110 print("\n CLUSTER ENFORCEMENT (live from the Kubernetes API)")
1111 for namespace, layer in sorted(enforcement.items()):
1112 status = layer.get("status", "unknown")
1113 if status != "ok":
1114 print(f" {namespace}: {status} — {layer.get('reason', 'no reason given')}")
1115 continue
1116 for name, hard in sorted(layer.get("resource_quotas", {}).items()):
1117 print(f" {namespace} ResourceQuota/{name}:")
1118 for key, value in sorted(hard.items()):
1119 print(f" {key}: {value}")
1120 for name, limits in sorted(layer.get("limit_ranges", {}).items()):
1121 print(f" {namespace} LimitRange/{name}:")
1122 for limit in limits:
1123 maximum = limit.get("max", {})
1124 if maximum:
1125 print(f" {limit.get('type', '?')} max: {maximum}")
1126 print()
1128 except Exception as e:
1129 formatter.print_error(f"Failed to get job validation policy: {e}")
1130 sys.exit(1)
1133# ---------------------------------------------------------------------------
1134# Policy pre-checks (advisory)
1135# ---------------------------------------------------------------------------
1138def _policy_regions(config: Any, requested: tuple[str, ...] | None) -> list[str]:
1139 """Regions to read policy from: those asked for, else the configured set."""
1140 if requested:
1141 seen: dict[str, None] = {}
1142 for region in requested:
1143 seen.setdefault(region, None)
1144 return list(seen)
1146 from ..status import _workload_regions, resolve_regions
1148 regions = _workload_regions(resolve_regions(config))
1149 return regions or [config.default_region]
1152def _render_verdicts(verdicts: Any, *, indent: str = " ") -> None:
1153 """Print one line per region plus its reasons, for a terminal reader."""
1154 from ..job_policy import VERDICT_ADMIT, VERDICT_REJECT, VERDICT_UNKNOWN
1156 marks = {VERDICT_ADMIT: "admit ", VERDICT_REJECT: "REJECT", VERDICT_UNKNOWN: " ? "}
1157 for verdict in verdicts:
1158 print(f"{indent}[{marks.get(verdict.verdict, '?')}] {verdict.region}")
1159 if verdict.verdict == VERDICT_UNKNOWN:
1160 print(f"{indent} policy unreadable: {verdict.reason}")
1161 continue
1162 for issue in verdict.issues:
1163 where = f"{issue.manifest} " if issue.manifest else ""
1164 print(f"{indent} {where}[{issue.check}] {issue.message}")
1165 if verdict.enforcement_gaps:
1166 print(
1167 f"{indent} note: live quota/LimitRange unreadable for "
1168 f"{', '.join(verdict.enforcement_gaps)} — only the front-door "
1169 f"caps were checked"
1170 )
1173def _run_pre_submit_policy_check(
1174 config: Any,
1175 job_manager: Any,
1176 formatter: Any,
1177 *,
1178 manifest_path: str,
1179 namespace: str | None,
1180 target_region: str | None,
1181) -> None:
1182 """Advisory pre-submit check against the target region's live policy.
1184 Deliberately non-blocking. The cluster is the authoritative gate and this
1185 reads a snapshot of its policy over the network, so a check that refused to
1186 submit on its own opinion would block valid jobs whenever it is stale or
1187 wrong. It prints what it found and returns.
1189 A failure to read the policy is also non-fatal for the same reason: not
1190 being able to check is not evidence of a problem, and the submission that
1191 follows would have happened anyway without the flag.
1192 """
1193 from ..job_policy import VERDICT_REJECT, fetch_region_policies, region_verdicts
1195 try:
1196 manifests = job_manager.load_manifests(manifest_path)
1197 # Match what the server will see: submit_job fills in the namespace for
1198 # manifests that do not declare one, and the namespace allowlist check
1199 # is against that resolved value.
1200 effective_namespace = namespace or config.default_namespace
1201 for manifest in manifests:
1202 if isinstance(manifest, dict):
1203 manifest.setdefault("metadata", {}).setdefault("namespace", effective_namespace)
1205 regions = [target_region] if target_region else _policy_regions(config, None)
1206 policies = fetch_region_policies(job_manager._aws_client, regions)
1207 verdicts = region_verdicts(manifests, policies)
1208 except Exception as e:
1209 formatter.print_warning(f"Policy pre-check could not run ({e}); submitting anyway")
1210 return
1212 rejecting = [v for v in verdicts if v.verdict == VERDICT_REJECT]
1213 if rejecting:
1214 formatter.print_warning(
1215 f"Policy pre-check: {len(rejecting)} of {len(verdicts)} region(s) would "
1216 f"reject these manifests. Submitting anyway (advisory)."
1217 )
1218 _render_verdicts(verdicts)
1219 else:
1220 readable = [v for v in verdicts if v.verdict != "unknown"]
1221 if readable:
1222 formatter.print_success(
1223 f"Policy pre-check: admissible in {', '.join(v.region for v in readable)}"
1224 )
1225 else:
1226 formatter.print_warning(
1227 "Policy pre-check: no region's policy could be read; submitting anyway"
1228 )
1229 _render_verdicts(verdicts)
1232def _cdk_job_validation_policy() -> tuple[dict[str, Any], str]:
1233 """Read ``context.job_validation_policy`` out of the local cdk.json.
1235 Returns the raw sub-document and the path it came from. Raises when there is
1236 no cdk.json to read, because silently checking against shipped defaults
1237 would look like a successful check of the user's configuration.
1238 """
1239 import json
1240 from pathlib import Path
1242 path = Path.cwd() / "cdk.json"
1243 if not path.is_file():
1244 raise FileNotFoundError(
1245 f"no cdk.json at {path}; --offline reads the policy from a checkout"
1246 )
1247 context = json.loads(path.read_text(encoding="utf-8")).get("context", {}) or {}
1248 policy = context.get("job_validation_policy", {}) or {}
1249 return policy, str(path)
1252def _check_policy_offline(
1253 config: Any,
1254 job_manager: Any,
1255 formatter: Any,
1256 *,
1257 manifest_path: str | None,
1258 namespace: str | None,
1259 fail_on_reject: bool,
1260) -> None:
1261 """Judge manifests against cdk.json, with no AWS calls.
1263 Strictly weaker than the online path and says so in its output. Two reasons
1264 it cannot be authoritative, both real rather than theoretical: a region may
1265 have been deployed from a different checkout of this file, and CDK appends
1266 the project's own ECR registry hostnames to ``trusted_registries`` at synth
1267 time, so a deployed region trusts registries that appear nowhere here. An
1268 image-provenance rejection offline is therefore a maybe, not a no.
1269 """
1270 import dataclasses
1272 from gco.job_admission import JobValidationPolicy
1274 from ..job_policy import evaluate_manifests
1276 if not manifest_path:
1277 formatter.print_error("--offline needs a MANIFEST_PATH to check")
1278 sys.exit(1)
1280 try:
1281 configured, source = _cdk_job_validation_policy()
1282 policy = JobValidationPolicy.from_cdk_context(configured)
1283 manifests = job_manager.load_manifests(manifest_path)
1284 effective_namespace = namespace or config.default_namespace
1285 for manifest in manifests:
1286 if isinstance(manifest, dict):
1287 manifest.setdefault("metadata", {}).setdefault("namespace", effective_namespace)
1288 issues = evaluate_manifests(manifests, policy)
1289 except Exception as e:
1290 formatter.print_error(f"Offline policy check failed: {e}")
1291 sys.exit(1)
1293 caveat = (
1294 "checked the CONFIGURED policy from cdk.json, not what any region has "
1295 "deployed; CDK also adds project ECR registries at synth time, so an "
1296 "image rejection here may pass in a real region"
1297 )
1299 if config.output_format != "table":
1300 formatter.print(
1301 {
1302 "source": source,
1303 "mode": "offline",
1304 "caveat": caveat,
1305 "admissible": not issues,
1306 "issues": [dataclasses.asdict(issue) for issue in issues],
1307 }
1308 )
1309 if fail_on_reject and issues:
1310 sys.exit(1)
1311 return
1313 print(f"\n Offline policy check — {source}")
1314 print(" " + "-" * 60)
1315 if issues:
1316 for issue in issues:
1317 where = f"{issue.manifest} " if issue.manifest else ""
1318 print(f" {where}[{issue.check}] {issue.message}")
1319 else:
1320 print(" no violations of the configured policy")
1321 print(f"\n Note: {caveat}")
1322 print()
1324 if fail_on_reject and issues:
1325 sys.exit(1)
1328@jobs.command("check-policy")
1329@click.argument("manifest_path", type=click.Path(exists=True), required=False)
1330@click.option(
1331 "--region",
1332 "-r",
1333 "regions",
1334 multiple=True,
1335 help="Region to check (repeatable). Defaults to every configured region.",
1336)
1337@click.option(
1338 "--namespace",
1339 "-n",
1340 help="Namespace to assume for manifests that don't declare their own",
1341)
1342@click.option(
1343 "--offline",
1344 is_flag=True,
1345 help=(
1346 "Check against cdk.json instead of calling AWS. Needs no credentials, "
1347 "but reports the CONFIGURED policy, not the deployed one"
1348 ),
1349)
1350@click.option(
1351 "--fail-on-reject",
1352 is_flag=True,
1353 help="Exit 1 when any checked region would reject (after printing)",
1354)
1355@pass_config
1356def check_policy(
1357 config: Any,
1358 manifest_path: Any,
1359 regions: Any,
1360 namespace: Any,
1361 offline: Any,
1362 fail_on_reject: Any,
1363) -> None:
1364 """Check which regions would admit a manifest, and compare their policies.
1366 Reads the policy each region actually enforces and evaluates the manifest
1367 against it with the same code the manifest processor runs. Two things this
1368 answers that submitting cannot:
1370 A job can be admissible in one region and over-cap in another. Without
1371 this you discover that by submitting and being rejected.
1373 There are no per-region policy overrides, so any field that differs
1374 between regions means a region was deployed from a different checkout.
1375 That stays invisible until a manifest that worked yesterday is refused.
1377 Omit MANIFEST_PATH to compare policies without judging anything.
1379 --offline answers the same question from cdk.json with no AWS calls, for
1380 pre-commit hooks and air-gapped checkouts. It is strictly weaker: it reports
1381 what the file configures, and a deployed region trusts ECR registries the
1382 file never mentions, so an image rejection may be a false positive.
1384 Advisory only: the cluster is the real gate and this reads a snapshot of
1385 its policy, so it exits 0 unless you pass --fail-on-reject.
1387 Examples:
1388 gco jobs check-policy examples/gpu-job.yaml
1389 gco jobs check-policy examples/gpu-job.yaml -r us-east-1 -r us-east-2
1390 gco jobs check-policy # policy comparison only
1391 gco -o json jobs check-policy examples/gpu-job.yaml
1392 gco jobs check-policy examples/gpu-job.yaml --offline
1393 """
1394 import dataclasses
1396 from ..job_policy import (
1397 VERDICT_REJECT,
1398 detect_policy_drift,
1399 ecr_augmentation,
1400 fetch_region_policies,
1401 region_verdicts,
1402 registry_drift,
1403 )
1405 formatter = get_output_formatter(config)
1406 job_manager = get_job_manager(config)
1408 if offline:
1409 _check_policy_offline(
1410 config,
1411 job_manager,
1412 formatter,
1413 manifest_path=manifest_path,
1414 namespace=namespace,
1415 fail_on_reject=fail_on_reject,
1416 )
1417 return
1419 try:
1420 target_regions = _policy_regions(config, regions)
1421 policies = fetch_region_policies(job_manager._aws_client, target_regions)
1423 manifests: list[dict[str, Any]] = []
1424 if manifest_path:
1425 manifests = job_manager.load_manifests(manifest_path)
1426 effective_namespace = namespace or config.default_namespace
1427 for manifest in manifests:
1428 if isinstance(manifest, dict):
1429 manifest.setdefault("metadata", {}).setdefault("namespace", effective_namespace)
1431 verdicts = region_verdicts(manifests, policies) if manifests else []
1432 drift = detect_policy_drift(policies)
1433 registries = registry_drift(policies)
1434 if registries is not None:
1435 drift = [*drift, registries]
1436 augmentation = ecr_augmentation(policies)
1437 except Exception as e:
1438 formatter.print_error(f"Failed to check policy: {e}")
1439 sys.exit(1)
1441 if config.output_format != "table":
1442 formatter.print(
1443 {
1444 "regions": target_regions,
1445 "unreadable": {entry.region: entry.reason for entry in policies if not entry.ok},
1446 "verdicts": [dataclasses.asdict(verdict) for verdict in verdicts],
1447 "policy_drift": [dataclasses.asdict(item) for item in drift],
1448 "ecr_augmentation": augmentation,
1449 }
1450 )
1451 if fail_on_reject and any(v.verdict == VERDICT_REJECT for v in verdicts):
1452 sys.exit(1)
1453 return
1455 if verdicts:
1456 print(f"\n Admissibility — {len(verdicts)} region(s)")
1457 print(" " + "-" * 60)
1458 _render_verdicts(verdicts)
1460 print(f"\n Cross-region policy agreement — {len(policies)} region(s) read")
1461 print(" " + "-" * 60)
1462 readable = [entry for entry in policies if entry.ok]
1463 if len(readable) < 2:
1464 print(" only one region readable; nothing to compare")
1465 elif not drift:
1466 print(f" identical across {', '.join(entry.region for entry in readable)}")
1467 else:
1468 print(" these fields differ, which means a region is running a")
1469 print(" different deployment of cdk.json than the others:")
1470 for item in drift:
1471 print(f" {item.field}:")
1472 for region, value in sorted(item.values.items()):
1473 print(f" {region}: {value}")
1475 added = {region: hosts for region, hosts in augmentation.items() if hosts}
1476 if added:
1477 print("\n ECR hostnames CDK added at synth time (absent from cdk.json)")
1478 print(" " + "-" * 60)
1479 for region, hosts in sorted(added.items()):
1480 for host in hosts:
1481 print(f" {region}: {host}")
1482 print()
1484 if fail_on_reject and any(v.verdict == VERDICT_REJECT for v in verdicts):
1485 sys.exit(1)
1488@jobs.command("submit-queue")
1489@click.argument("manifest_path", type=click.Path(exists=True))
1490@click.option("--region", "-r", required=True, help="Target region for job execution")
1491@click.option("--namespace", "-n", default="gco-jobs", help="Kubernetes namespace")
1492@click.option(
1493 "--priority",
1494 "-p",
1495 type=click.IntRange(0, 100),
1496 default=0,
1497 help="Job priority (0-100, higher = more important)",
1498)
1499@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
1500@pass_config
1501def submit_job_queue(
1502 config: Any, manifest_path: Any, region: Any, namespace: Any, priority: Any, label: Any
1503) -> None:
1504 """Submit a job to the global DynamoDB queue for regional pickup.
1506 Jobs are stored in DynamoDB and picked up by the target region's
1507 manifest processor. This enables global job submission with
1508 centralized tracking and status history.
1510 This is different from submit-sqs which uses regional SQS queues.
1511 The DynamoDB queue provides:
1512 - Global visibility of all queued jobs
1513 - Status tracking and history
1514 - Priority-based scheduling
1515 - Cross-region job management
1517 Use 'gco queue list' to view queued jobs and their status.
1519 Examples:
1520 gco jobs submit-queue job.yaml --region us-east-1
1521 gco jobs submit-queue job.yaml -r us-west-2 --priority 50
1522 gco jobs submit-queue job.yaml -r us-east-1 -l team=ml -l project=training
1523 """
1525 from gco.services.manifest_processor import safe_load_yaml
1527 formatter = get_output_formatter(config)
1529 # Parse and validate labels at the CLI boundary. Silently dropping a
1530 # malformed value would submit a differently labelled workload than the
1531 # operator requested; values may still contain additional ``=`` bytes.
1532 labels = {}
1533 for lbl in label:
1534 key, separator, value = lbl.partition("=")
1535 if not separator or not key:
1536 raise click.BadParameter(
1537 "labels must use key=value with a non-empty key",
1538 param_hint="--label",
1539 )
1540 labels[key] = value
1542 try:
1543 # Load manifest
1544 with open(manifest_path, encoding="utf-8") as f:
1545 manifest = safe_load_yaml(f, allow_aliases=False)
1547 # Submit via API
1548 from ..aws_client import get_aws_client
1550 aws_client = get_aws_client(config)
1552 result = aws_client.call_api(
1553 method="POST",
1554 path="/api/v1/queue/jobs",
1555 region=region,
1556 body={
1557 "manifest": manifest,
1558 "target_region": region,
1559 "namespace": namespace,
1560 "priority": priority,
1561 "labels": labels if labels else None,
1562 },
1563 )
1565 formatter.print_success(f"Job queued for {region}")
1566 formatter.print_info("Use 'gco queue list' or 'gco queue get <job_id>' to track status")
1567 formatter.print(result)
1569 except Exception as e:
1570 formatter.print_error(f"Failed to queue job: {e}")
1571 sys.exit(1)