Coverage for cli / commands / status_cmd.py: 100.00%
165 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Fleet-wide status command for GCO."""
3from __future__ import annotations
5import sys
6import time
7from typing import Any
9import click
11from ..config import GCOConfig
12from ..output import get_output_formatter
13from ..status import (
14 COST_REFRESH_INTERVAL_SECONDS,
15 SECTION_CAPACITY,
16 SECTION_COSTS,
17 SECTION_INFERENCE,
18 SECTION_JOBS,
19 SECTION_NODEPOOLS,
20 SECTION_ORDER,
21 SECTION_QUEUE,
22 SECTION_REGIONS,
23 SECTION_STACKS,
24 SEVERITY_ERROR,
25 STATUS_EMPTY,
26 STATUS_OK,
27 WATCH_INTERVAL_FLOOR_SECONDS,
28 FleetStatus,
29 Section,
30 gather_fleet_status,
31)
33pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
35# The summary is a route into the existing per-domain commands; every block
36# names its drill-down so the reader never has to guess the next command.
37_DRILL_DOWNS = {
38 SECTION_REGIONS: "gco stacks regions list",
39 SECTION_STACKS: "gco stacks status <name> --region <region>",
40 SECTION_QUEUE: "gco queue stats",
41 SECTION_JOBS: "gco jobs list --all-regions",
42 SECTION_CAPACITY: "gco capacity status",
43 SECTION_INFERENCE: "gco inference list",
44 SECTION_COSTS: "gco costs summary",
45 SECTION_NODEPOOLS: "gco nodepools list",
46}
49def _render_regions(data: dict[str, Any]) -> list[str]:
50 workload = ", ".join(data.get("workload", [])) or "-"
51 return [
52 f"global: {data.get('global')} api-gateway: {data.get('api_gateway')} "
53 f"monitoring: {data.get('monitoring')}",
54 f"workload: {workload} (source: {data.get('source')})",
55 ]
58def _render_stacks(data: dict[str, Any]) -> list[str]:
59 lines = []
60 entries = list(data.get("expected", []))
61 optional = list(data.get("optional", []))
62 width = max((len(str(e.get("name"))) for e in entries + optional), default=0)
63 for entry, marker in [(e, "") for e in entries] + [(e, " (optional)") for e in optional]:
64 status_text = entry.get("status") or "-"
65 lines.append(
66 f"{str(entry.get('name')):<{width}} {status_text:<25} "
67 f"{str(entry.get('health')):<12} {entry.get('region')}{marker}"
68 )
69 return lines
72def _render_queue(data: dict[str, Any]) -> list[str]:
73 lines = []
74 by_region = data.get("by_region", {})
75 for region, entry in by_region.items():
76 dlq = entry.get("dlq")
77 dlq_text = "unknown" if dlq is None else str(dlq)
78 lines.append(
79 f"{region:<15} available {entry.get('available', 0):<5} "
80 f"in-flight {entry.get('in_flight', 0):<5} delayed {entry.get('delayed', 0):<5} "
81 f"dlq {dlq_text}"
82 )
83 totals = data.get("totals", {})
84 if by_region and totals:
85 lines.append(
86 f"{'totals':<15} available {totals.get('available', 0):<5} "
87 f"in-flight {totals.get('in_flight', 0):<5} delayed {totals.get('delayed', 0):<5} "
88 f"dlq {totals.get('dlq', 0)}"
89 )
90 return lines
93def _render_jobs(data: dict[str, Any]) -> list[str]:
94 totals = data.get("totals", {})
95 complete = data.get("complete", True)
96 scan = "complete" if complete else f"TRUNCATED after {data.get('records_evaluated')} records"
97 lines = [
98 f"total {totals.get('total', 0)} queued {totals.get('queued', 0)} "
99 f"running {totals.get('running', 0)} (scan {scan})"
100 ]
101 for region, statuses in data.get("by_region", {}).items():
102 counts = " ".join(f"{key} {value}" for key, value in statuses.items())
103 lines.append(f"{region:<15} {counts}")
104 return lines
107def _render_capacity(data: dict[str, Any]) -> list[str]:
108 lines = []
109 for region, entry in data.get("by_region", {}).items():
110 signals = ", ".join(entry.get("unavailable_signals", []))
111 missing = f" (unavailable: {signals})" if signals else ""
112 lines.append(
113 f"{region:<15} queue {entry.get('queue_depth', 0):<4} "
114 f"running {entry.get('running_jobs', 0):<4} "
115 f"gpu {entry.get('gpu_utilization', 0.0):<5.1f} "
116 f"cpu {entry.get('cpu_utilization', 0.0):<5.1f} "
117 f"telemetry {entry.get('telemetry_status')}{missing}"
118 )
119 return lines
122def _render_inference(data: dict[str, Any]) -> list[str]:
123 totals = data.get("totals", {})
124 summary = ", ".join(f"{state} {count}" for state, count in totals.items()) or "none"
125 lines = [f"endpoints: {data.get('count', 0)} ({summary})"]
126 for endpoint in data.get("endpoints", []):
127 regions = ",".join(endpoint.get("target_regions") or [])
128 lines.append(
129 f"{endpoint.get('endpoint_name')} {endpoint.get('desired_state')} "
130 f"regions [{regions}] namespace {endpoint.get('namespace')}"
131 )
132 return lines
135def _render_costs(data: dict[str, Any]) -> list[str]:
136 lines = []
137 if "total" in data:
138 lines.append(f"total ${data.get('total', 0.0):.2f} over {data.get('window_days')} days")
139 for item in data.get("by_service", []):
140 lines.append(f"{item.get('service', ''):<40} ${item.get('amount', 0.0):.2f}")
141 tags = data.get("allocation_tags")
142 if tags is not None:
143 rendered_tags = ", ".join(
144 f"{item.get('tag_key', '')}={item.get('status', 'unknown')}" for item in tags
145 )
146 lines.append(f"cost allocation tags: {rendered_tags or 'none'}")
147 if data.get("as_of"):
148 lines.append(f"as of {data['as_of']}")
149 return lines
152def _render_nodepools(data: dict[str, Any]) -> list[str]:
153 lines = []
154 for region, entry in data.get("by_region", {}).items():
155 pools = entry.get("nodepools")
156 if pools is None:
157 lines.append(f"{region:<15} {entry.get('note', 'endpoint not reachable')}")
158 else:
159 names = ", ".join(str(p.get("name")) for p in pools) or "none"
160 lines.append(f"{region:<15} {len(pools)} nodepools: {names}")
161 return lines
164_SECTION_BODIES = {
165 SECTION_REGIONS: _render_regions,
166 SECTION_STACKS: _render_stacks,
167 SECTION_QUEUE: _render_queue,
168 SECTION_JOBS: _render_jobs,
169 SECTION_CAPACITY: _render_capacity,
170 SECTION_INFERENCE: _render_inference,
171 SECTION_COSTS: _render_costs,
172 SECTION_NODEPOOLS: _render_nodepools,
173}
176def _render_section(section: Section) -> list[str]:
177 """One compact block: heading with status and drill-down, then detail."""
178 heading = f"{section.name} [{section.status}]"
179 drill = _DRILL_DOWNS.get(section.name)
180 if drill:
181 heading = f"{heading} · {drill}"
182 lines = [heading]
184 if section.status in (STATUS_OK, STATUS_EMPTY):
185 render_body = _SECTION_BODIES.get(section.name)
186 body = render_body(section.data) if render_body else []
187 if section.status == STATUS_EMPTY and not body:
188 body = ["nothing here — the read succeeded and found no records"]
189 lines.extend(f" {line}" for line in body)
190 else:
191 # Skipped and unavailable sections show their reason instead of
192 # silently vanishing from the summary.
193 lines.append(f" {section.reason or 'no reason recorded'}")
194 for error in section.errors[:5]:
195 lines.append(f" error: {error}")
196 remaining = len(section.errors) - 5
197 if remaining > 0:
198 lines.append(f" ... and {remaining} more error(s)")
199 return lines
202def _render_table(doc: FleetStatus) -> None:
203 """Render the document for a terminal reader.
205 The nested document would collapse to ``<dict>`` cells inside the
206 generic table formatter, so this renderer is hand-rolled: verdict
207 first, findings second, then one block per section.
208 """
209 print(f"Fleet status: {doc.overall.upper()} project {doc.project_name}")
210 print(f"generated {doc.generated_at}")
211 if doc.degraded:
212 print(f"degraded sections: {', '.join(doc.degraded)}")
214 print()
215 if doc.findings:
216 print("Findings:")
217 for finding in doc.findings:
218 print(f" [{finding.severity}] {finding.section}: {finding.message}")
219 else:
220 print("Findings: none — nothing looks wrong.")
222 for name in SECTION_ORDER:
223 section = doc.sections.get(name)
224 if section is None:
225 continue
226 print()
227 for line in _render_section(section):
228 print(line)
231def _has_error_finding(doc: FleetStatus) -> bool:
232 return any(finding.severity == SEVERITY_ERROR for finding in doc.findings)
235def _watch_loop(
236 config: GCOConfig,
237 *,
238 region: str | None,
239 with_costs: bool,
240 with_nodepools: bool,
241 interval: int,
242 fail_on_findings: bool,
243 with_policy: bool = False,
244) -> None:
245 """Re-gather and redraw until interrupted.
247 The costs section is re-fetched at most once per
248 ``COST_REFRESH_INTERVAL_SECONDS``; ticks in between reuse the previous
249 section, whose ``as_of`` timestamp shows when the figure was actually
250 retrieved. The reuse is in-process only — no cache file is written.
251 """
252 cached_costs: Section | None = None
253 cached_costs_at = 0.0
254 while True:
255 now = time.monotonic()
256 reuse = None
257 if (
258 with_costs
259 and cached_costs is not None
260 and now - cached_costs_at < COST_REFRESH_INTERVAL_SECONDS
261 ):
262 reuse = cached_costs
263 doc = gather_fleet_status(
264 config,
265 region=region,
266 with_costs=with_costs,
267 with_nodepools=with_nodepools,
268 with_policy=with_policy,
269 costs_cache=reuse,
270 )
271 if with_costs and reuse is None:
272 cached_costs = doc.sections.get(SECTION_COSTS)
273 cached_costs_at = now
275 click.clear()
276 _render_table(doc)
277 if fail_on_findings and _has_error_finding(doc):
278 sys.exit(1)
279 print(f"\nrefreshing every {interval}s — Ctrl-C to stop")
280 time.sleep(interval)
283@click.command("status")
284@click.option("--region", "-r", help="Restrict the gather to a single region")
285@click.option(
286 "--with-costs",
287 is_flag=True,
288 help="Include the costs section (Cost Explorer bills per request)",
289)
290@click.option(
291 "--with-nodepools",
292 is_flag=True,
293 help="Include Karpenter nodepools (requires a reachable cluster API endpoint)",
294)
295@click.option(
296 "--watch",
297 type=int,
298 default=None,
299 metavar="SECONDS",
300 help=(
301 f"Re-gather and redraw every SECONDS (minimum "
302 f"{WATCH_INTERVAL_FLOOR_SECONDS}; table output only)"
303 ),
304)
305@click.option(
306 "--with-policy",
307 is_flag=True,
308 help=(
309 "Compare the job-validation policy each region enforces and report any "
310 "field that differs (one API call per region)"
311 ),
312)
313@click.option(
314 "--fail-on-findings",
315 is_flag=True,
316 help=(
317 "Exit 1 when any error-severity finding is present (after rendering); "
318 "with --watch, exits on the first tick that carries one"
319 ),
320)
321@pass_config
322def status(
323 config: GCOConfig,
324 region: str | None,
325 with_costs: bool,
326 with_nodepools: bool,
327 with_policy: bool,
328 watch: int | None,
329 fail_on_findings: bool,
330) -> None:
331 """Show fleet-wide deployment status across configured regions.
333 Aggregates control-plane state — stacks, queue depth, jobs, capacity,
334 and inference endpoints — into one document. Every section carries its
335 own status, so a failed read degrades that section instead of hiding
336 the rest. Reads that bill per request or need cluster reachability are
337 opt-in flags.
339 Examples:
340 gco status
341 gco status -r us-east-1
342 gco status --output json
343 gco status --with-costs --with-nodepools
344 gco status --with-policy
345 gco status --watch 10
346 gco status --fail-on-findings
347 """
348 formatter = get_output_formatter(config)
350 if watch is not None:
351 if config.output_format != "table":
352 formatter.print_error(
353 "--watch requires table output; a repeating stream of documents "
354 "is not consumable as JSON or YAML"
355 )
356 sys.exit(1)
357 if watch < WATCH_INTERVAL_FLOOR_SECONDS:
358 formatter.print_error(
359 f"--watch interval must be at least {WATCH_INTERVAL_FLOOR_SECONDS} seconds"
360 )
361 sys.exit(1)
362 try:
363 _watch_loop(
364 config,
365 region=region,
366 with_costs=with_costs,
367 with_nodepools=with_nodepools,
368 with_policy=with_policy,
369 interval=watch,
370 fail_on_findings=fail_on_findings,
371 )
372 except KeyboardInterrupt:
373 return
374 return
376 doc = gather_fleet_status(
377 config,
378 region=region,
379 with_costs=with_costs,
380 with_nodepools=with_nodepools,
381 with_policy=with_policy,
382 )
384 if config.output_format == "table":
385 _render_table(doc)
386 else:
387 formatter.print(doc)
389 if fail_on_findings and _has_error_finding(doc):
390 sys.exit(1)