Coverage for cli/commands/queue_cmd.py: 87.18%
186 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Global job queue commands."""
3import sys
4from typing import Any
6import click
8from ..config import GCOConfig
9from ..output import get_output_formatter
11pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
14@click.group()
15@pass_config
16def queue(config: Any) -> None:
17 """Manage the global job queue (DynamoDB-backed).
19 The job queue provides centralized job submission and tracking:
20 - Submit jobs to any region from anywhere
21 - Track job status globally
22 - View job history and statistics
23 """
24 pass
27@queue.command("submit")
28@click.argument("manifest_path", type=click.Path(exists=True))
29@click.option("--region", "-r", required=True, help="Target region for job execution")
30@click.option("--namespace", "-n", default="gco-jobs", help="Kubernetes namespace")
31@click.option("--priority", "-p", default=0, help="Job priority (0-100, higher = more important)")
32@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
33@click.option(
34 "--max-spot-price",
35 type=float,
36 help=(
37 "Spot price cap in USD/hour. The job is held in the queue until the "
38 "current spot price of --spot-instance-type in the target region "
39 "drops to or below this value. Requires --spot-instance-type."
40 ),
41)
42@click.option(
43 "--spot-instance-type",
44 help=(
45 "EC2 instance type whose spot price gates dispatch (e.g. g5.xlarge). "
46 "Requires --max-spot-price."
47 ),
48)
49@pass_config
50def queue_submit(
51 config: Any,
52 manifest_path: Any,
53 region: Any,
54 namespace: Any,
55 priority: Any,
56 label: Any,
57 max_spot_price: Any,
58 spot_instance_type: Any,
59) -> None:
60 """Submit a job to the global queue for regional pickup.
62 Jobs are stored in DynamoDB and picked up by the target region's
63 manifest processor. This enables global job submission with
64 centralized tracking.
66 With --max-spot-price and --spot-instance-type the job is cost-gated:
67 it stays queued until spot pricing for that instance type in the target
68 region drops to or below the cap. Cancel with `gco queue cancel` if the
69 price never clears.
71 Examples:
72 gco queue submit job.yaml --region us-east-1
73 gco queue submit job.yaml -r us-west-2 --priority 50
74 gco queue submit job.yaml -r us-east-1 -l team=ml -l project=training
75 gco queue submit job.yaml -r us-east-1 --max-spot-price 0.50 --spot-instance-type g5.xlarge
76 """
78 from gco.services.manifest_processor import safe_load_yaml
79 from gco.services.spot_price_gate import validate_spot_gate_fields
81 formatter = get_output_formatter(config)
83 gate_error = validate_spot_gate_fields(max_spot_price, spot_instance_type)
84 if gate_error: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 formatter.print_error(gate_error)
86 sys.exit(1)
88 # Parse labels
89 labels = {}
90 for lbl in label:
91 if "=" in lbl: 91 ↛ 90line 91 didn't jump to line 90 because the condition on line 91 was always true
92 k, v = lbl.split("=", 1)
93 labels[k] = v
95 try:
96 # Load manifest
97 with open(manifest_path, encoding="utf-8") as f:
98 manifest = safe_load_yaml(f, allow_aliases=False)
100 # Submit via API
101 from ..aws_client import get_aws_client
103 aws_client = get_aws_client(config)
105 body = {
106 "manifest": manifest,
107 "target_region": region,
108 "namespace": namespace,
109 "priority": priority,
110 "labels": labels if labels else None,
111 }
112 if max_spot_price is not None: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 body["max_spot_price"] = max_spot_price
114 body["spot_instance_type"] = spot_instance_type
116 result = aws_client.call_api(
117 method="POST",
118 path="/api/v1/queue/jobs",
119 region=region if config.use_regional_api else None,
120 body=body,
121 )
123 formatter.print_success(f"Job queued for {region}")
124 if max_spot_price is not None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 formatter.print_info(
126 f"Spot price gate: dispatches when {spot_instance_type} spot "
127 f"price in {region} is <= ${max_spot_price}/hour"
128 )
129 formatter.print(result)
131 except Exception as e:
132 formatter.print_error(f"Failed to queue job: {e}")
133 sys.exit(1)
136@queue.command("list")
137@click.option("--region", "-r", help="Filter by target region")
138@click.option(
139 "--status",
140 "-s",
141 type=click.Choice(["queued", "claimed", "running", "succeeded", "failed", "cancelled"]),
142 help="Filter by status",
143)
144@click.option("--namespace", "-n", help="Filter by namespace")
145@click.option("--limit", "-l", default=50, help="Maximum results")
146@pass_config
147def queue_list(config: Any, region: Any, status: Any, namespace: Any, limit: Any) -> None:
148 """List jobs in the global queue.
150 Examples:
151 gco queue list
152 gco queue list --region us-east-1 --status queued
153 gco queue list -s running
154 """
155 formatter = get_output_formatter(config)
157 try:
158 from ..aws_client import get_aws_client
160 aws_client = get_aws_client(config)
162 # Build query params
163 params = {"limit": limit}
164 if region:
165 params["target_region"] = region
166 if status:
167 params["status"] = status
168 if namespace:
169 params["namespace"] = namespace
171 # The region is a DynamoDB filter, not a transport pin. The global API
172 # can serve it; forced regional mode uses the configured default bridge.
173 query_region = config.default_region if config.use_regional_api else None
174 result = aws_client.call_api(
175 method="GET",
176 path="/api/v1/queue/jobs",
177 region=query_region,
178 params=params,
179 )
181 if config.output_format == "table": 181 ↛ 200line 181 didn't jump to line 200 because the condition on line 181 was always true
182 jobs = result.get("jobs", [])
183 if not jobs:
184 formatter.print_info("No jobs found")
185 return
187 print(f"\n Queued Jobs ({result.get('count', 0)} total)")
188 print(" " + "-" * 90)
189 print(
190 " JOB ID NAME REGION STATUS"
191 )
192 print(" " + "-" * 90)
193 for job in jobs:
194 job_id = job.get("job_id", "")[:36]
195 name = job.get("job_name", "")[:22]
196 target = job.get("target_region", "")[:14]
197 job_status = job.get("status", "")[:10]
198 print(f" {job_id:<36} {name:<23} {target:<15} {job_status}")
199 else:
200 formatter.print(result)
202 except Exception as e:
203 formatter.print_error(f"Failed to list queued jobs: {e}")
204 sys.exit(1)
207@queue.command("get")
208@click.argument("job_id")
209@click.option("--region", "-r", help="Region to query (any region works)")
210@pass_config
211def queue_get(config: Any, job_id: Any, region: Any) -> None:
212 """Get details of a queued job including status history.
214 Examples:
215 gco queue get abc123-def456
216 gco queue get abc123-def456 --region us-east-1
217 """
218 formatter = get_output_formatter(config)
220 try:
221 from ..aws_client import get_aws_client
223 aws_client = get_aws_client(config)
225 query_region = region or (config.default_region if config.use_regional_api else None)
226 result = aws_client.call_api(
227 method="GET",
228 path=f"/api/v1/queue/jobs/{job_id}",
229 region=query_region,
230 )
232 job = result.get("job", {})
234 if config.output_format == "table": 234 ↛ 270line 234 didn't jump to line 270 because the condition on line 234 was always true
235 print(f"\n Job: {job.get('job_id')}")
236 print(" " + "-" * 50)
237 print(f" Name: {job.get('job_name')}")
238 print(f" Target Region: {job.get('target_region')}")
239 print(f" Namespace: {job.get('namespace')}")
240 print(f" Status: {job.get('status')}")
241 print(f" Priority: {job.get('priority')}")
242 print(f" Submitted: {job.get('submitted_at')}")
243 if job.get("spot_max_price"): 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 print(
245 f" Spot Gate: {job.get('spot_instance_type')} <= "
246 f"${job.get('spot_max_price')}/hour"
247 )
248 if job.get("spot_gate_observed_price"):
249 print(
250 f" Last Price: ${job.get('spot_gate_observed_price')} "
251 f"(checked {job.get('spot_gate_checked_at')})"
252 )
253 if job.get("claimed_by"): 253 ↛ 255line 253 didn't jump to line 255 because the condition on line 253 was always true
254 print(f" Claimed By: {job.get('claimed_by')}")
255 if job.get("completed_at"): 255 ↛ 256line 255 didn't jump to line 256 because the condition on line 255 was never true
256 print(f" Completed: {job.get('completed_at')}")
257 if job.get("error_message"): 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 print(f" Error: {job.get('error_message')}")
260 # Show status history
261 history = job.get("status_history", [])
262 if history: 262 ↛ exitline 262 didn't return from function 'queue_get' because the condition on line 262 was always true
263 print("\n Status History:")
264 for entry in history:
265 ts = entry.get("timestamp", "")[:19]
266 st = entry.get("status", "")
267 msg = entry.get("message", "")[:40]
268 print(f" [{ts}] {st}: {msg}")
269 else:
270 formatter.print(result)
272 except Exception as e:
273 formatter.print_error(f"Failed to get job: {e}")
274 sys.exit(1)
277@queue.command("cancel")
278@click.argument("job_id")
279@click.option("--reason", help="Cancellation reason")
280@click.option("--region", "-r", help="Region to query (any region works)")
281@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
282@pass_config
283def queue_cancel(config: Any, job_id: Any, reason: Any, region: Any, yes: Any) -> None:
284 """Cancel a queued job (only works for jobs not yet running).
286 Examples:
287 gco queue cancel abc123-def456
288 gco queue cancel abc123-def456 --reason "No longer needed"
289 """
290 formatter = get_output_formatter(config)
292 if not yes: 292 ↛ 293line 292 didn't jump to line 293 because the condition on line 292 was never true
293 click.confirm(f"Cancel job {job_id}?", abort=True)
295 try:
296 from ..aws_client import get_aws_client
298 aws_client = get_aws_client(config)
300 query_region = region or (config.default_region if config.use_regional_api else None)
301 params = {}
302 if reason:
303 params["reason"] = reason
305 result = aws_client.call_api(
306 method="DELETE",
307 path=f"/api/v1/queue/jobs/{job_id}",
308 region=query_region,
309 params=params,
310 )
312 formatter.print_success(f"Job {job_id} cancelled")
313 formatter.print(result)
315 except Exception as e:
316 formatter.print_error(f"Failed to cancel job: {e}")
317 sys.exit(1)
320@queue.command("stats")
321@click.option("--region", "-r", help="Region to query (any region works)")
322@pass_config
323def queue_stats(config: Any, region: Any) -> None:
324 """Get job queue statistics by region and status.
326 Examples:
327 gco queue stats
328 """
329 formatter = get_output_formatter(config)
331 try:
332 from ..aws_client import get_aws_client
334 aws_client = get_aws_client(config)
336 query_region = region or (config.default_region if config.use_regional_api else None)
337 result = aws_client.call_api(
338 method="GET",
339 path="/api/v1/queue/stats",
340 region=query_region,
341 )
343 if config.output_format == "table": 343 ↛ 364line 343 didn't jump to line 364 because the condition on line 343 was always true
344 summary = result.get("summary", {})
345 by_region = result.get("by_region", {})
347 print("\n Job Queue Statistics")
348 print(" " + "-" * 50)
349 print(f" Total Jobs: {summary.get('total_jobs', 0)}")
350 print(f" Queued: {summary.get('total_queued', 0)}")
351 print(f" Running: {summary.get('total_running', 0)}")
353 if by_region: 353 ↛ exitline 353 didn't return from function 'queue_stats' because the condition on line 353 was always true
354 print("\n By Region:")
355 print(" REGION QUEUED RUNNING SUCCEEDED FAILED")
356 print(" " + "-" * 55)
357 for reg, statuses in by_region.items():
358 queued = statuses.get("queued", 0)
359 running = statuses.get("running", 0)
360 succeeded = statuses.get("succeeded", 0)
361 failed = statuses.get("failed", 0)
362 print(f" {reg:<15} {queued:>6} {running:>7} {succeeded:>9} {failed:>6}")
363 else:
364 formatter.print(result)
366 except Exception as e:
367 formatter.print_error(f"Failed to get queue stats: {e}")
368 sys.exit(1)