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