Coverage for cli / commands / queue_cmd.py: 100.00%
190 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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 """
38 pass
41@queue.command("submit")
42@click.argument("manifest_path", type=click.Path(exists=True))
43@click.option("--region", "-r", required=True, help="Target region for job execution")
44@click.option("--namespace", "-n", default="gco-jobs", help="Kubernetes namespace")
45@click.option("--priority", "-p", default=0, help="Job priority (0-100, higher = more important)")
46@click.option("--label", "-l", multiple=True, help="Add labels (key=value)")
47@click.option(
48 "--max-spot-price",
49 type=float,
50 help=(
51 "Spot price cap in USD/hour. The job is held in the queue until the "
52 "current spot price of --spot-instance-type in the target region "
53 "drops to or below this value. Requires --spot-instance-type."
54 ),
55)
56@click.option(
57 "--spot-instance-type",
58 help=(
59 "EC2 instance type whose spot price gates dispatch (e.g. g5.xlarge). "
60 "Requires --max-spot-price."
61 ),
62)
63@pass_config
64def queue_submit(
65 config: Any,
66 manifest_path: Any,
67 region: Any,
68 namespace: Any,
69 priority: Any,
70 label: Any,
71 max_spot_price: Any,
72 spot_instance_type: Any,
73) -> None:
74 """Submit a job to the global queue for regional pickup.
76 Jobs are stored in DynamoDB and picked up by the target region's
77 manifest processor. This enables global job submission with
78 centralized tracking.
80 With --max-spot-price and --spot-instance-type the job is cost-gated:
81 it stays queued until spot pricing for that instance type in the target
82 region drops to or below the cap. Cancel with `gco queue cancel` if the
83 price never clears.
85 Examples:
86 gco queue submit job.yaml --region us-east-1
87 gco queue submit job.yaml -r us-west-2 --priority 50
88 gco queue submit job.yaml -r us-east-1 -l team=ml -l project=training
89 gco queue submit job.yaml -r us-east-1 --max-spot-price 0.50 --spot-instance-type g5.xlarge
90 """
92 from gco.services.manifest_processor import safe_load_yaml
93 from gco.services.spot_price_gate import validate_spot_gate_fields
95 formatter = get_output_formatter(config)
97 gate_error = validate_spot_gate_fields(max_spot_price, spot_instance_type)
98 if gate_error:
99 formatter.print_error(gate_error)
100 sys.exit(1)
102 # Parse labels before opening the manifest or constructing an AWS client.
103 labels = _parse_labels(tuple(label))
105 try:
106 # Load manifest
107 with open(manifest_path, encoding="utf-8") as f:
108 manifest = safe_load_yaml(f, allow_aliases=False)
110 # Submit via API
111 from ..aws_client import get_aws_client
113 aws_client = get_aws_client(config)
115 body = {
116 "manifest": manifest,
117 "target_region": region,
118 "namespace": namespace,
119 "priority": priority,
120 "labels": labels if labels else None,
121 }
122 if max_spot_price is not None:
123 body["max_spot_price"] = max_spot_price
124 body["spot_instance_type"] = spot_instance_type
126 result = aws_client.call_api(
127 method="POST",
128 path="/api/v1/queue/jobs",
129 region=region if config.use_regional_api else None,
130 body=body,
131 )
133 formatter.print_success(f"Job queued for {region}")
134 if max_spot_price is not None:
135 formatter.print_info(
136 f"Spot price gate: dispatches when {spot_instance_type} spot "
137 f"price in {region} is <= ${max_spot_price}/hour"
138 )
139 formatter.print(result)
141 except Exception as e:
142 formatter.print_error(f"Failed to queue job: {e}")
143 sys.exit(1)
146@queue.command("list")
147@click.option("--region", "-r", help="Filter by target region")
148@click.option(
149 "--status",
150 "-s",
151 type=click.Choice(["queued", "claimed", "running", "succeeded", "failed", "cancelled"]),
152 help="Filter by status",
153)
154@click.option("--namespace", "-n", help="Filter by namespace")
155@click.option("--limit", "-l", default=50, help="Maximum results")
156@pass_config
157def queue_list(config: Any, region: Any, status: Any, namespace: Any, limit: Any) -> None:
158 """List jobs in the global queue.
160 Examples:
161 gco queue list
162 gco queue list --region us-east-1 --status queued
163 gco queue list -s running
164 """
165 formatter = get_output_formatter(config)
167 try:
168 from ..aws_client import get_aws_client
170 aws_client = get_aws_client(config)
172 # Build query params
173 params = {"limit": limit}
174 if region:
175 params["target_region"] = region
176 if status:
177 params["status"] = status
178 if namespace:
179 params["namespace"] = namespace
181 # The region is a DynamoDB filter, not a transport pin. The global API
182 # can serve it; forced regional mode uses the configured default bridge.
183 query_region = config.default_region if config.use_regional_api else None
184 result = aws_client.call_api(
185 method="GET",
186 path="/api/v1/queue/jobs",
187 region=query_region,
188 params=params,
189 )
191 if config.output_format == "table":
192 jobs = result.get("jobs", [])
193 if not jobs:
194 formatter.print_info("No jobs found")
195 return
197 print(f"\n Queued Jobs ({result.get('count', 0)} total)")
198 print(" " + "-" * 90)
199 print(
200 " JOB ID NAME REGION STATUS"
201 )
202 print(" " + "-" * 90)
203 for job in jobs:
204 job_id = job.get("job_id", "")[:36]
205 name = job.get("job_name", "")[:22]
206 target = job.get("target_region", "")[:14]
207 job_status = job.get("status", "")[:10]
208 print(f" {job_id:<36} {name:<23} {target:<15} {job_status}")
209 else:
210 formatter.print(result)
212 except Exception as e:
213 formatter.print_error(f"Failed to list queued jobs: {e}")
214 sys.exit(1)
217@queue.command("get")
218@click.argument("job_id")
219@click.option("--region", "-r", help="Region to query (any region works)")
220@pass_config
221def queue_get(config: Any, job_id: Any, region: Any) -> None:
222 """Get details of a queued job including status history.
224 Examples:
225 gco queue get abc123-def456
226 gco queue get abc123-def456 --region us-east-1
227 """
228 formatter = get_output_formatter(config)
230 try:
231 from ..aws_client import get_aws_client
233 aws_client = get_aws_client(config)
235 query_region = region or (config.default_region if config.use_regional_api else None)
236 result = aws_client.call_api(
237 method="GET",
238 path=f"/api/v1/queue/jobs/{job_id}",
239 region=query_region,
240 )
242 job = result.get("job", {})
244 if config.output_format == "table":
245 print(f"\n Job: {job.get('job_id')}")
246 print(" " + "-" * 50)
247 print(f" Name: {job.get('job_name')}")
248 print(f" Target Region: {job.get('target_region')}")
249 print(f" Namespace: {job.get('namespace')}")
250 print(f" Status: {job.get('status')}")
251 print(f" Priority: {job.get('priority')}")
252 print(f" Submitted: {job.get('submitted_at')}")
253 if job.get("spot_max_price"):
254 print(
255 f" Spot Gate: {job.get('spot_instance_type')} <= "
256 f"${job.get('spot_max_price')}/hour"
257 )
258 if job.get("spot_gate_observed_price"):
259 print(
260 f" Last Price: ${job.get('spot_gate_observed_price')} "
261 f"(checked {job.get('spot_gate_checked_at')})"
262 )
263 if job.get("claimed_by"):
264 print(f" Claimed By: {job.get('claimed_by')}")
265 if job.get("completed_at"):
266 print(f" Completed: {job.get('completed_at')}")
267 if job.get("error_message"):
268 print(f" Error: {job.get('error_message')}")
270 # Show status history
271 history = job.get("status_history", [])
272 if history:
273 print("\n Status History:")
274 for entry in history:
275 ts = entry.get("timestamp", "")[:19]
276 st = entry.get("status", "")
277 msg = entry.get("message", "")[:40]
278 print(f" [{ts}] {st}: {msg}")
279 else:
280 formatter.print(result)
282 except Exception as e:
283 formatter.print_error(f"Failed to get job: {e}")
284 sys.exit(1)
287@queue.command("cancel")
288@click.argument("job_id")
289@click.option("--reason", help="Cancellation reason")
290@click.option("--region", "-r", help="Region to query (any region works)")
291@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
292@pass_config
293def queue_cancel(config: Any, job_id: Any, reason: Any, region: Any, yes: Any) -> None:
294 """Cancel a queued job (only works for jobs not yet running).
296 Examples:
297 gco queue cancel abc123-def456
298 gco queue cancel abc123-def456 --reason "No longer needed"
299 """
300 formatter = get_output_formatter(config)
302 if not yes:
303 confirm(f"Cancel job {job_id}?", abort=True)
305 try:
306 from ..aws_client import get_aws_client
308 aws_client = get_aws_client(config)
310 query_region = region or (config.default_region if config.use_regional_api else None)
311 params = {}
312 if reason:
313 params["reason"] = reason
315 result = aws_client.call_api(
316 method="DELETE",
317 path=f"/api/v1/queue/jobs/{job_id}",
318 region=query_region,
319 params=params,
320 )
322 formatter.print_success(f"Job {job_id} cancelled")
323 formatter.print(result)
325 except Exception as e:
326 formatter.print_error(f"Failed to cancel job: {e}")
327 sys.exit(1)
330@queue.command("stats")
331@click.option("--region", "-r", help="Region to query (any region works)")
332@pass_config
333def queue_stats(config: Any, region: Any) -> None:
334 """Get job queue statistics by region and status.
336 Examples:
337 gco queue stats
338 """
339 formatter = get_output_formatter(config)
341 try:
342 from ..aws_client import get_aws_client
344 aws_client = get_aws_client(config)
346 query_region = region or (config.default_region if config.use_regional_api else None)
347 result = aws_client.call_api(
348 method="GET",
349 path="/api/v1/queue/stats",
350 region=query_region,
351 )
353 if config.output_format == "table":
354 summary = result.get("summary", {})
355 by_region = result.get("by_region", {})
357 print("\n Job Queue Statistics")
358 print(" " + "-" * 50)
359 print(f" Total Jobs: {summary.get('total_jobs', 0)}")
360 print(f" Queued: {summary.get('total_queued', 0)}")
361 print(f" Running: {summary.get('total_running', 0)}")
363 if by_region:
364 print("\n By Region:")
365 print(" REGION QUEUED RUNNING SUCCEEDED FAILED")
366 print(" " + "-" * 55)
367 for reg, statuses in by_region.items():
368 queued = statuses.get("queued", 0)
369 running = statuses.get("running", 0)
370 succeeded = statuses.get("succeeded", 0)
371 failed = statuses.get("failed", 0)
372 print(f" {reg:<15} {queued:>6} {running:>7} {succeeded:>9} {failed:>6}")
373 else:
374 formatter.print(result)
376 except Exception as e:
377 formatter.print_error(f"Failed to get queue stats: {e}")
378 sys.exit(1)