Coverage for cli / commands / queue_cmd.py: 100.00%
189 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Global job queue commands."""
3import sys
4from typing import Any
6import click
8from ..config import GCOConfig
9from ..output import confirm, get_output_formatter
11pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
14def _parse_labels(raw_labels: tuple[str, ...]) -> dict[str, str]:
15 """Parse repeated ``--label KEY=VALUE`` options without dropping bad input."""
16 labels: dict[str, str] = {}
17 for raw_label in raw_labels:
18 key, separator, value = raw_label.partition("=")
19 if not separator or not key or not value:
20 raise click.BadParameter(
21 "must be KEY=VALUE with a non-empty key and value",
22 param_hint="--label",
23 )
24 labels[key] = value
25 return labels
28@click.group()
29@pass_config
30def queue(config: Any) -> None:
31 """Manage the global job queue (DynamoDB-backed).
33 The job queue provides centralized job submission and tracking:
34 - Submit jobs to any region from anywhere
35 - Track job status globally
36 - View job history and statistics
37 """
40@queue.command("submit")
41@click.argument("manifest_path", type=click.Path(exists=True))
42@click.option("--region", "-r", required=True, help="Target region for job execution")
43@click.option("--namespace", "-n", default="gco-jobs", help="Kubernetes namespace")
44@click.option("--priority", "-p", default=0, help="Job priority (0-100, higher = more important)")
45@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
46@click.option(
47 "--max-spot-price",
48 type=float,
49 help=(
50 "Spot price cap in USD/hour. The job is held in the queue until the "
51 "current spot price of --spot-instance-type in the target region "
52 "drops to or below this value. Requires --spot-instance-type."
53 ),
54)
55@click.option(
56 "--spot-instance-type",
57 help=(
58 "EC2 instance type whose spot price gates dispatch (e.g. g5.xlarge). "
59 "Requires --max-spot-price."
60 ),
61)
62@pass_config
63def queue_submit(
64 config: Any,
65 manifest_path: Any,
66 region: Any,
67 namespace: Any,
68 priority: Any,
69 label: Any,
70 max_spot_price: Any,
71 spot_instance_type: Any,
72) -> None:
73 """Submit a job to the global queue for regional pickup.
75 Jobs are stored in DynamoDB and picked up by the target region's
76 manifest processor. This enables global job submission with
77 centralized tracking.
79 With --max-spot-price and --spot-instance-type the job is cost-gated:
80 it stays queued until spot pricing for that instance type in the target
81 region drops to or below the cap. Cancel with `gco queue cancel` if the
82 price never clears.
84 Examples:
85 gco queue submit job.yaml --region us-east-1
86 gco queue submit job.yaml -r us-west-2 --priority 50
87 gco queue submit job.yaml -r us-east-1 -l team=ml -l project=training
88 gco queue submit job.yaml -r us-east-1 --max-spot-price 0.50 --spot-instance-type g5.xlarge
89 """
91 from gco.services.manifest_processor import safe_load_yaml
92 from gco.services.spot_price_gate import validate_spot_gate_fields
94 formatter = get_output_formatter(config)
96 gate_error = validate_spot_gate_fields(max_spot_price, spot_instance_type)
97 if gate_error:
98 formatter.print_error(gate_error)
99 sys.exit(1)
101 # Parse labels before opening the manifest or constructing an AWS client.
102 labels = _parse_labels(tuple(label))
104 try:
105 # Load manifest
106 with open(manifest_path, encoding="utf-8") as f:
107 manifest = safe_load_yaml(f, allow_aliases=False)
109 # Submit via API
110 from ..aws_client import get_aws_client
112 aws_client = get_aws_client(config)
114 body = {
115 "manifest": manifest,
116 "target_region": region,
117 "namespace": namespace,
118 "priority": priority,
119 "labels": labels if labels else None,
120 }
121 if max_spot_price is not None:
122 body["max_spot_price"] = max_spot_price
123 body["spot_instance_type"] = spot_instance_type
125 result = aws_client.call_api(
126 method="POST",
127 path="/api/v1/queue/jobs",
128 region=region if config.use_regional_api else None,
129 body=body,
130 )
132 formatter.print_success(f"Job queued for {region}")
133 if max_spot_price is not None:
134 formatter.print_info(
135 f"Spot price gate: dispatches when {spot_instance_type} spot "
136 f"price in {region} is <= ${max_spot_price}/hour"
137 )
138 formatter.print(result)
140 except Exception as e:
141 formatter.print_error(f"Failed to queue job: {e}")
142 sys.exit(1)
145@queue.command("list")
146@click.option("--region", "-r", help="Filter by target region")
147@click.option(
148 "--status",
149 "-s",
150 type=click.Choice(["queued", "claimed", "running", "succeeded", "failed", "cancelled"]),
151 help="Filter by status",
152)
153@click.option("--namespace", "-n", help="Filter by namespace")
154@click.option("--limit", "-l", default=50, help="Maximum results")
155@pass_config
156def queue_list(config: Any, region: Any, status: Any, namespace: Any, limit: Any) -> None:
157 """List jobs in the global queue.
159 Examples:
160 gco queue list
161 gco queue list --region us-east-1 --status queued
162 gco queue list -s running
163 """
164 formatter = get_output_formatter(config)
166 try:
167 from ..aws_client import get_aws_client
169 aws_client = get_aws_client(config)
171 # Build query params
172 params = {"limit": limit}
173 if region:
174 params["target_region"] = region
175 if status:
176 params["status"] = status
177 if namespace:
178 params["namespace"] = namespace
180 # The region is a DynamoDB filter, not a transport pin. The global API
181 # can serve it; forced regional mode uses the configured default bridge.
182 query_region = config.default_region if config.use_regional_api else None
183 result = aws_client.call_api(
184 method="GET",
185 path="/api/v1/queue/jobs",
186 region=query_region,
187 params=params,
188 )
190 if config.output_format == "table":
191 jobs = result.get("jobs", [])
192 if not jobs:
193 formatter.print_info("No jobs found")
194 return
196 print(f"\n Queued Jobs ({result.get('count', 0)} total)")
197 print(" " + "-" * 90)
198 print(
199 " JOB ID NAME REGION STATUS"
200 )
201 print(" " + "-" * 90)
202 for job in jobs:
203 job_id = job.get("job_id", "")[:36]
204 name = job.get("job_name", "")[:22]
205 target = job.get("target_region", "")[:14]
206 job_status = job.get("status", "")[:10]
207 print(f" {job_id:<36} {name:<23} {target:<15} {job_status}")
208 else:
209 formatter.print(result)
211 except Exception as e:
212 formatter.print_error(f"Failed to list queued jobs: {e}")
213 sys.exit(1)
216@queue.command("get")
217@click.argument("job_id")
218@click.option("--region", "-r", help="Region to query (any region works)")
219@pass_config
220def queue_get(config: Any, job_id: Any, region: Any) -> None:
221 """Get details of a queued job including status history.
223 Examples:
224 gco queue get abc123-def456
225 gco queue get abc123-def456 --region us-east-1
226 """
227 formatter = get_output_formatter(config)
229 try:
230 from ..aws_client import get_aws_client
232 aws_client = get_aws_client(config)
234 query_region = region or (config.default_region if config.use_regional_api else None)
235 result = aws_client.call_api(
236 method="GET",
237 path=f"/api/v1/queue/jobs/{job_id}",
238 region=query_region,
239 )
241 job = result.get("job", {})
243 if config.output_format == "table":
244 print(f"\n Job: {job.get('job_id')}")
245 print(" " + "-" * 50)
246 print(f" Name: {job.get('job_name')}")
247 print(f" Target Region: {job.get('target_region')}")
248 print(f" Namespace: {job.get('namespace')}")
249 print(f" Status: {job.get('status')}")
250 print(f" Priority: {job.get('priority')}")
251 print(f" Submitted: {job.get('submitted_at')}")
252 if job.get("spot_max_price"):
253 print(
254 f" Spot Gate: {job.get('spot_instance_type')} <= "
255 f"${job.get('spot_max_price')}/hour"
256 )
257 if job.get("spot_gate_observed_price"):
258 print(
259 f" Last Price: ${job.get('spot_gate_observed_price')} "
260 f"(checked {job.get('spot_gate_checked_at')})"
261 )
262 if job.get("claimed_by"):
263 print(f" Claimed By: {job.get('claimed_by')}")
264 if job.get("completed_at"):
265 print(f" Completed: {job.get('completed_at')}")
266 if job.get("error_message"):
267 print(f" Error: {job.get('error_message')}")
269 # Show status history
270 history = job.get("status_history", [])
271 if history:
272 print("\n Status History:")
273 for entry in history:
274 ts = entry.get("timestamp", "")[:19]
275 st = entry.get("status", "")
276 msg = entry.get("message", "")[:40]
277 print(f" [{ts}] {st}: {msg}")
278 else:
279 formatter.print(result)
281 except Exception as e:
282 formatter.print_error(f"Failed to get job: {e}")
283 sys.exit(1)
286@queue.command("cancel")
287@click.argument("job_id")
288@click.option("--reason", help="Cancellation reason")
289@click.option("--region", "-r", help="Region to query (any region works)")
290@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
291@pass_config
292def queue_cancel(config: Any, job_id: Any, reason: Any, region: Any, yes: Any) -> None:
293 """Cancel a queued job (only works for jobs not yet running).
295 Examples:
296 gco queue cancel abc123-def456
297 gco queue cancel abc123-def456 --reason "No longer needed"
298 """
299 formatter = get_output_formatter(config)
301 if not yes:
302 confirm(f"Cancel job {job_id}?", abort=True)
304 try:
305 from ..aws_client import get_aws_client
307 aws_client = get_aws_client(config)
309 query_region = region or (config.default_region if config.use_regional_api else None)
310 params = {}
311 if reason:
312 params["reason"] = reason
314 result = aws_client.call_api(
315 method="DELETE",
316 path=f"/api/v1/queue/jobs/{job_id}",
317 region=query_region,
318 params=params,
319 )
321 formatter.print_success(f"Job {job_id} cancelled")
322 formatter.print(result)
324 except Exception as e:
325 formatter.print_error(f"Failed to cancel job: {e}")
326 sys.exit(1)
329@queue.command("stats")
330@click.option("--region", "-r", help="Region to query (any region works)")
331@pass_config
332def queue_stats(config: Any, region: Any) -> None:
333 """Get job queue statistics by region and status.
335 Examples:
336 gco queue stats
337 """
338 formatter = get_output_formatter(config)
340 try:
341 from ..aws_client import get_aws_client
343 aws_client = get_aws_client(config)
345 query_region = region or (config.default_region if config.use_regional_api else None)
346 result = aws_client.call_api(
347 method="GET",
348 path="/api/v1/queue/stats",
349 region=query_region,
350 )
352 if config.output_format == "table":
353 summary = result.get("summary", {})
354 by_region = result.get("by_region", {})
356 print("\n Job Queue Statistics")
357 print(" " + "-" * 50)
358 print(f" Total Jobs: {summary.get('total_jobs', 0)}")
359 print(f" Queued: {summary.get('total_queued', 0)}")
360 print(f" Running: {summary.get('total_running', 0)}")
362 if by_region:
363 print("\n By Region:")
364 print(" REGION QUEUED RUNNING SUCCEEDED FAILED")
365 print(" " + "-" * 55)
366 for reg, statuses in by_region.items():
367 queued = statuses.get("queued", 0)
368 running = statuses.get("running", 0)
369 succeeded = statuses.get("succeeded", 0)
370 failed = statuses.get("failed", 0)
371 print(f" {reg:<15} {queued:>6} {running:>7} {succeeded:>9} {failed:>6}")
372 else:
373 formatter.print(result)
375 except Exception as e:
376 formatter.print_error(f"Failed to get queue stats: {e}")
377 sys.exit(1)