Coverage for cli / status.py: 100.00%
440 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Fleet-wide status document assembly for ``gco status``.
3Gathers control-plane state across the configured deployment regions and
4returns it as a single :class:`FleetStatus` document. Every section is
5gathered independently behind a uniform envelope and carries its own status,
6so a failure in one section never suppresses the rest, and "there is nothing
7there" is always distinguishable from "the read could not be performed".
9The document — not any one rendering of it — is the public contract:
10``gco status --output json`` emits it unchanged, and the ``fleet_status``
11MCP tool returns that JSON to agents.
12"""
14from __future__ import annotations
16import logging
17import threading
18import time
19from collections import Counter
20from collections.abc import Callable
21from concurrent.futures import ThreadPoolExecutor
22from dataclasses import dataclass, field
23from datetime import UTC, datetime
24from typing import TYPE_CHECKING, Any
26from botocore.exceptions import ClientError
28from cli.config import GCOConfig, _load_cdk_json
30if TYPE_CHECKING:
31 from cli.capacity.multi_region import RegionCapacity
32 from cli.stacks import StackInfo
34logger = logging.getLogger(__name__)
36# ---------------------------------------------------------------------------
37# Status vocabulary
38# ---------------------------------------------------------------------------
39# Module-level constants rather than bare strings at call sites, so a typo is
40# an AttributeError instead of a silently wrong status.
42#: The read succeeded and returned data.
43STATUS_OK = "ok"
44#: The read succeeded and there is genuinely nothing — a success, not a
45#: degradation.
46STATUS_EMPTY = "empty"
47#: The read succeeded for some regions or signals and failed for others.
48STATUS_PARTIAL = "partial"
49#: The read could not be attempted, for a known and explainable reason.
50STATUS_UNAVAILABLE = "unavailable"
51#: The read was attempted and failed unexpectedly.
52STATUS_ERROR = "error"
53#: The section was not requested (an opt-in section without its flag).
54STATUS_SKIPPED = "skipped"
56#: Finding severities.
57SEVERITY_ERROR = "error"
58SEVERITY_WARN = "warn"
60#: Overall document verdicts.
61OVERALL_OK = "ok"
62OVERALL_DEGRADED = "degraded"
64# Section names. The tuple fixes the rendering and JSON key order.
65SECTION_REGIONS = "regions"
66SECTION_STACKS = "stacks"
67SECTION_QUEUE = "queue"
68SECTION_JOBS = "jobs"
69SECTION_CAPACITY = "capacity"
70SECTION_INFERENCE = "inference"
71SECTION_COSTS = "costs"
72SECTION_NODEPOOLS = "nodepools"
73SECTION_POLICY = "policy"
75SECTION_ORDER: tuple[str, ...] = (
76 SECTION_REGIONS,
77 SECTION_STACKS,
78 SECTION_QUEUE,
79 SECTION_JOBS,
80 SECTION_CAPACITY,
81 SECTION_INFERENCE,
82 SECTION_COSTS,
83 SECTION_NODEPOOLS,
84 SECTION_POLICY,
85)
87# Sections that fan out over the resolved workload region list and therefore
88# cannot be gathered at all when region resolution fails.
89_PER_REGION_SECTIONS = frozenset({SECTION_STACKS, SECTION_QUEUE, SECTION_CAPACITY})
91# Section statuses that degrade the overall document. ``skipped`` is absent
92# on purpose: skipping is what the operator asked for.
93_DEGRADED_STATUSES = frozenset({STATUS_PARTIAL, STATUS_UNAVAILABLE, STATUS_ERROR})
95# Source markers for the ``regions`` section, so a reader can tell a
96# configured topology from a flag-narrowed one.
97REGION_SOURCE_CDK_JSON = "cdk.json"
98REGION_SOURCE_FLAG = "--region flag"
100# Stack health classification. Anything ending ``_IN_PROGRESS`` is a deploy
101# in flight; ``UPDATE_ROLLBACK_COMPLETE`` and friends mean the last deploy
102# did not take, which is unhealthy even though CloudFormation is at rest.
103_HEALTHY_STACK_STATUSES = frozenset(
104 {
105 "CREATE_COMPLETE",
106 "UPDATE_COMPLETE",
107 "IMPORT_COMPLETE",
108 }
109)
110_IN_PROGRESS_SUFFIX = "_IN_PROGRESS"
112HEALTH_HEALTHY = "healthy"
113HEALTH_IN_PROGRESS = "in-progress"
114HEALTH_NOT_DEPLOYED = "not-deployed"
115HEALTH_UNHEALTHY = "unhealthy"
117#: Wall-clock budget for one section's gather; a section that exceeds it
118#: reports ``error`` instead of holding the whole document.
119SECTION_TIMEOUT_SECONDS = 30
121#: Cost Explorer window for the opt-in ``costs`` section.
122COST_WINDOW_DAYS = 30
124#: Minimum ``--watch`` interval, so watch mode cannot hammer AWS APIs.
125WATCH_INTERVAL_FLOOR_SECONDS = 5
127#: Minimum spacing between Cost Explorer fetches under ``--watch``.
128#: Cost Explorer bills per request and its data does not change minute to
129#: minute; the in-between ticks reuse the last section, whose ``as_of``
130#: shows when the figure was actually retrieved.
131COST_REFRESH_INTERVAL_SECONDS = 15 * 60
133#: Ceiling for concurrent per-region (or per-stack) reads within a section.
134_MAX_FANOUT_WORKERS = 8
137# ---------------------------------------------------------------------------
138# Document model
139# ---------------------------------------------------------------------------
142@dataclass(frozen=True)
143class Section:
144 """One independently gathered part of the status document."""
146 name: str
147 status: str
148 data: dict[str, Any] = field(default_factory=dict)
149 reason: str | None = None
150 errors: list[str] = field(default_factory=list)
153@dataclass(frozen=True)
154class Finding:
155 """Something that looks wrong, derived from an already-gathered document."""
157 severity: str
158 section: str
159 message: str
162@dataclass(frozen=True)
163class FleetStatus:
164 """The whole fleet status document."""
166 generated_at: str
167 project_name: str
168 overall: str
169 degraded: list[str]
170 findings: list[Finding]
171 sections: dict[str, Section]
174# ---------------------------------------------------------------------------
175# Region topology resolution
176# ---------------------------------------------------------------------------
178_REGIONS_UNAVAILABLE_REASON = (
179 "deployment regions are not configured; run from a checkout containing "
180 "cdk.json (context.deployment_regions.regional) or pass --region to name "
181 "one explicitly"
182)
185def resolve_regions(config: GCOConfig, region: str | None = None) -> Section:
186 """Resolve the deployment topology the rest of the document fans out over.
188 Reads ``context.deployment_regions`` from ``cdk.json``. An explicit
189 ``region`` narrows the workload list to exactly that region and records
190 the narrowing in ``source``. This function never falls back to scanning
191 AWS regions for stacks — an unresolvable topology is reported as
192 ``unavailable`` instead of being guessed at.
193 """
194 cdk_regions = _load_cdk_json()
195 configured = [item for item in cdk_regions.get("regional", []) if isinstance(item, str)]
197 if region:
198 workload = [region]
199 source = REGION_SOURCE_FLAG
200 elif configured:
201 workload = configured
202 source = REGION_SOURCE_CDK_JSON
203 else:
204 return Section(
205 name=SECTION_REGIONS,
206 status=STATUS_UNAVAILABLE,
207 reason=_REGIONS_UNAVAILABLE_REASON,
208 )
210 return Section(
211 name=SECTION_REGIONS,
212 status=STATUS_OK,
213 data={
214 "global": cdk_regions.get("global", config.global_region),
215 "api_gateway": cdk_regions.get("api_gateway", config.api_gateway_region),
216 "monitoring": cdk_regions.get("monitoring", config.monitoring_region),
217 "workload": workload,
218 "source": source,
219 },
220 )
223def _workload_regions(regions_section: Section) -> list[str]:
224 """Return the resolved workload region list, or ``[]`` when unresolved."""
225 if regions_section.status != STATUS_OK:
226 return []
227 workload = regions_section.data.get("workload", [])
228 return [item for item in workload if isinstance(item, str)]
231# ---------------------------------------------------------------------------
232# Section gathering
233# ---------------------------------------------------------------------------
236def _run_section(name: str, gather: Callable[[], Section]) -> Section:
237 """Run one section gatherer, absorbing any escaping exception.
239 This boundary is what makes "a failure in one section never suppresses
240 the rest" structural: no gatherer exception can reach the renderer.
241 """
242 try:
243 return gather()
244 except Exception as e:
245 logger.debug("Status section %s failed: %s", name, e)
246 return Section(
247 name=name,
248 status=STATUS_ERROR,
249 reason="the read failed unexpectedly",
250 errors=[f"{type(e).__name__}: {e}"],
251 )
254def _run_sections_concurrently(gatherers: dict[str, Callable[[], Section]]) -> dict[str, Section]:
255 """Run section gatherers in parallel under the shared wall-clock budget.
257 Every gatherer gets the full :data:`SECTION_TIMEOUT_SECONDS` of wall
258 clock because they run concurrently. A section that has not finished by
259 the deadline reports ``error`` naming the timeout. Sections run on
260 daemon threads so an abandoned straggler — a hung subprocess probe or
261 an unresponsive endpoint — can neither hold the document nor block
262 process exit afterwards.
263 """
264 results: dict[str, Section] = {}
265 lock = threading.Lock()
267 def run(name: str, gather: Callable[[], Section]) -> None:
268 section = _run_section(name, gather)
269 with lock:
270 results[name] = section
272 threads = {
273 name: threading.Thread(target=run, args=(name, gather), name=f"status-{name}", daemon=True)
274 for name, gather in gatherers.items()
275 }
276 for thread in threads.values():
277 thread.start()
278 deadline = time.monotonic() + SECTION_TIMEOUT_SECONDS
279 for thread in threads.values():
280 thread.join(max(0.0, deadline - time.monotonic()))
282 sections: dict[str, Section] = {}
283 with lock:
284 for name in gatherers:
285 sections[name] = results.get(name) or Section(
286 name=name,
287 status=STATUS_ERROR,
288 reason=f"the gather exceeded the {SECTION_TIMEOUT_SECONDS}s section timeout",
289 )
290 return sections
293def _regions_unavailable_section(name: str) -> Section:
294 """Section placeholder used when the workload region list is unresolved."""
295 return Section(
296 name=name,
297 status=STATUS_UNAVAILABLE,
298 reason="deployment regions could not be resolved; see the regions section",
299 )
302def _fanout_workers(count: int) -> int:
303 """Bounded worker count for a per-region or per-stack fan-out."""
304 return max(1, min(_MAX_FANOUT_WORKERS, count))
307def _probe_regional_stacks(config: GCOConfig, workload: list[str]) -> dict[str, StackInfo | None]:
308 """Describe each workload region's regional stack directly.
310 This single probe round feeds the ``stacks`` section and gates the
311 ``queue`` and ``capacity`` gathers: both delegate to managers that fall
312 back to scanning every AWS region when stack discovery comes up empty,
313 and this command must never trigger that scan. ``None`` means the stack
314 is absent or not readable — CloudFormation reads here never raise.
315 """
316 from cli.stacks import get_stack_manager
318 manager = get_stack_manager(config)
320 def probe(region: str) -> StackInfo | None:
321 return manager.get_stack_status(f"{config.regional_stack_prefix}-{region}", region)
323 with ThreadPoolExecutor(max_workers=_fanout_workers(len(workload))) as pool:
324 return dict(zip(workload, pool.map(probe, workload), strict=True))
327# ---------------------------------------------------------------------------
328# stacks
329# ---------------------------------------------------------------------------
332def _classify_stack_health(status: str | None) -> str:
333 """Classify a CloudFormation stack status for the document."""
334 if status is None:
335 return HEALTH_NOT_DEPLOYED
336 if status in _HEALTHY_STACK_STATUSES:
337 return HEALTH_HEALTHY
338 if status.endswith(_IN_PROGRESS_SUFFIX):
339 return HEALTH_IN_PROGRESS
340 return HEALTH_UNHEALTHY
343def _stack_entry(name: str, region: str, info: StackInfo | None) -> dict[str, Any]:
344 """One stack's document entry; ``info`` is None when absent or unreadable."""
345 status = info.status if info else None
346 updated = info.updated_time.isoformat() if info and info.updated_time else None
347 return {
348 "name": name,
349 "region": region,
350 "status": status,
351 "health": _classify_stack_health(status),
352 "updated_time": updated,
353 }
356def _gather_stacks(
357 config: GCOConfig,
358 regions_data: dict[str, Any],
359 workload: list[str],
360) -> Section:
361 """Describe every expected and optional stack of the deployment.
363 Expected stacks are the global, API-gateway, and monitoring stacks plus
364 one regional stack per workload region. The per-region API bridges and
365 the analytics stack are optional — not deploying them is a valid
366 configuration — so they are listed only when present and their absence
367 never produces a finding.
368 """
369 from cli.stacks import get_stack_manager
371 project = config.project_name
372 global_region = str(regions_data.get("global", config.global_region))
373 api_gateway_region = str(regions_data.get("api_gateway", config.api_gateway_region))
374 monitoring_region = str(regions_data.get("monitoring", config.monitoring_region))
376 expected: list[tuple[str, str]] = [
377 (config.global_stack_name, global_region),
378 (config.api_gateway_stack_name, api_gateway_region),
379 (f"{project}-monitoring", monitoring_region),
380 ]
381 expected.extend((f"{config.regional_stack_prefix}-{region}", region) for region in workload)
382 optional: list[tuple[str, str]] = [
383 (f"{project}-regional-api-{region}", region) for region in workload
384 ]
385 optional.append((f"{project}-analytics", api_gateway_region))
387 manager = get_stack_manager(config)
388 everything = expected + optional
390 def describe(spec: tuple[str, str]) -> StackInfo | None:
391 return manager.get_stack_status(spec[0], spec[1])
393 with ThreadPoolExecutor(max_workers=_fanout_workers(len(everything))) as pool:
394 described = dict(
395 zip([name for name, _ in everything], pool.map(describe, everything), strict=True)
396 )
398 expected_entries = [
399 _stack_entry(name, region, described.get(name)) for name, region in expected
400 ]
401 optional_entries = [
402 _stack_entry(name, region, described[name])
403 for name, region in optional
404 if described.get(name) is not None
405 ]
407 return Section(
408 name=SECTION_STACKS,
409 status=STATUS_OK,
410 data={"expected": expected_entries, "optional": optional_entries},
411 )
414# ---------------------------------------------------------------------------
415# queue
416# ---------------------------------------------------------------------------
419def _gather_queue(
420 config: GCOConfig,
421 workload: list[str],
422 regional_probe: dict[str, StackInfo | None],
423 checkout_configured: bool,
424) -> Section:
425 """Read job-queue and dead-letter-queue depth per workload region.
427 Regions whose regional stack is absent are reported ``unavailable``
428 without attempting the read: the queue lookup rediscovers stacks
429 internally and must only run when the fast discovery path is guaranteed
430 to succeed.
431 """
432 from cli.jobs import get_job_manager
434 manager = get_job_manager(config)
435 by_region: dict[str, dict[str, int | None]] = {}
436 errors: list[str] = []
437 unavailable = 0
439 def read(region: str) -> dict[str, Any] | None:
440 return manager.get_queue_status(region)
442 prefix = config.regional_stack_prefix
443 readable = [region for region in workload if regional_probe.get(region) is not None]
444 for region in workload:
445 if region not in readable:
446 unavailable += 1
447 errors.append(f"{region}: regional stack {prefix}-{region} is absent or not readable")
448 if readable and not checkout_configured:
449 unavailable += len(readable)
450 errors.extend(
451 f"{region}: queue reads need the configured region list from cdk.json"
452 for region in readable
453 )
454 readable = []
456 results: dict[str, dict[str, Any] | Exception] = {}
457 if readable:
458 with ThreadPoolExecutor(max_workers=_fanout_workers(len(readable))) as pool:
459 futures = {region: pool.submit(read, region) for region in readable}
460 for region, future in futures.items():
461 try:
462 results[region] = future.result() or {}
463 except Exception as e:
464 results[region] = e
466 unexpected = 0
467 for region in readable:
468 outcome = results[region]
469 if isinstance(outcome, ValueError):
470 unavailable += 1
471 errors.append(f"{region}: {outcome}")
472 elif isinstance(outcome, Exception):
473 unexpected += 1
474 errors.append(f"{region}: {type(outcome).__name__}: {outcome}")
475 else:
476 by_region[region] = {
477 "available": int(outcome.get("messages_available", 0)),
478 "in_flight": int(outcome.get("messages_in_flight", 0)),
479 "delayed": int(outcome.get("messages_delayed", 0)),
480 "dlq": outcome.get("dlq_messages"),
481 }
483 totals = {
484 "available": sum(entry["available"] or 0 for entry in by_region.values()),
485 "in_flight": sum(entry["in_flight"] or 0 for entry in by_region.values()),
486 "delayed": sum(entry["delayed"] or 0 for entry in by_region.values()),
487 "dlq": sum(entry["dlq"] or 0 for entry in by_region.values()),
488 }
489 data = {"by_region": by_region, "totals": totals}
491 if by_region and not errors:
492 return Section(name=SECTION_QUEUE, status=STATUS_OK, data=data)
493 if by_region:
494 reason = f"queue depth unavailable for {len(workload) - len(by_region)} of {len(workload)} regions"
495 return Section(
496 name=SECTION_QUEUE, status=STATUS_PARTIAL, data=data, reason=reason, errors=errors
497 )
498 if unexpected:
499 return Section(
500 name=SECTION_QUEUE,
501 status=STATUS_ERROR,
502 data=data,
503 reason="the job queue could not be read in any workload region",
504 errors=errors,
505 )
506 return Section(
507 name=SECTION_QUEUE,
508 status=STATUS_UNAVAILABLE,
509 data=data,
510 reason=(
511 f"no readable job queue in any workload region; deploy the regional "
512 f"stack(s) with `gco stacks deploy {prefix}-<region>`"
513 ),
514 errors=errors,
515 )
518# ---------------------------------------------------------------------------
519# jobs
520# ---------------------------------------------------------------------------
523def _gather_jobs(config: GCOConfig, region: str | None) -> Section:
524 """Read fleet-wide job counts from the queue-statistics API route."""
525 from cli.aws_client import get_aws_client
527 aws_client = get_aws_client(config)
528 query_region = region or (config.default_region if config.use_regional_api else None)
529 try:
530 # A status snapshot reports a failing route honestly instead of
531 # retrying through it; the next gather re-reads anyway, and retries
532 # here can outlive the section timeout.
533 result = aws_client.call_api(
534 method="GET", path="/api/v1/queue/stats", region=query_region, max_attempts=1
535 )
536 except RuntimeError as e:
537 if "API endpoint" not in str(e):
538 raise
539 stack = config.api_gateway_stack_name
540 return Section(
541 name=SECTION_JOBS,
542 status=STATUS_UNAVAILABLE,
543 reason=(f"the {stack} API is unreachable; deploy it with `gco stacks deploy {stack}`"),
544 errors=[str(e)],
545 )
547 summary = result.get("summary") or {}
548 by_region = result.get("by_region") or {}
549 totals = {
550 "total": int(summary.get("total_jobs", 0)),
551 "queued": int(summary.get("total_queued", 0)),
552 "running": int(summary.get("total_running", 0)),
553 }
554 data: dict[str, Any] = {
555 "totals": totals,
556 "by_region": by_region,
557 "complete": bool(summary.get("complete", True)),
558 "records_evaluated": summary.get("records_evaluated"),
559 }
560 status = STATUS_EMPTY if not by_region and totals["total"] == 0 else STATUS_OK
561 return Section(name=SECTION_JOBS, status=status, data=data)
564# ---------------------------------------------------------------------------
565# capacity
566# ---------------------------------------------------------------------------
569def _capacity_entry_from(cap: RegionCapacity) -> dict[str, Any]:
570 """Document entry for one region's capacity sweep result."""
571 return {
572 "queue_depth": int(cap.queue_depth),
573 "running_jobs": int(cap.running_jobs),
574 "gpu_utilization": float(cap.gpu_utilization),
575 "cpu_utilization": float(cap.cpu_utilization),
576 "telemetry_status": str(cap.telemetry_status),
577 "unavailable_signals": list(cap.unavailable_signals),
578 }
581def _capacity_unavailable_entry() -> dict[str, Any]:
582 """Entry for a region whose capacity telemetry could not be attempted."""
583 return {
584 "queue_depth": 0,
585 "running_jobs": 0,
586 "gpu_utilization": 0.0,
587 "cpu_utilization": 0.0,
588 "telemetry_status": STATUS_UNAVAILABLE,
589 "unavailable_signals": ["queue", "gpu", "cpu"],
590 }
593def _gather_capacity(
594 config: GCOConfig,
595 workload: list[str],
596 configured: list[str],
597 regional_probe: dict[str, StackInfo | None],
598) -> Section:
599 """Read per-region queue depth and utilization with telemetry provenance.
601 The multi-region checker rediscovers stacks internally and falls back to
602 scanning every AWS region when discovery finds nothing, so it is invoked
603 only when at least one probed regional stack exists and the configured
604 region list is non-empty — conditions under which the fast discovery
605 path always succeeds. Regions failing that gate get an honest
606 ``unavailable`` telemetry entry instead.
607 """
608 from cli.capacity import get_multi_region_capacity_checker
610 prefix = config.regional_stack_prefix
611 by_region: dict[str, dict[str, Any]] = {}
612 errors: list[str] = []
614 present = [region for region in workload if regional_probe.get(region) is not None]
615 sweep_returned_empty = False
616 if present and configured:
617 checker = get_multi_region_capacity_checker(config)
618 if set(workload) == set(configured):
619 capacities = checker.get_all_regions_capacity()
620 # Sweep-level failures are what make an empty result mean
621 # "checks failed" rather than "no regions".
622 errors.extend(checker._last_region_errors)
623 sweep_returned_empty = not capacities
624 for cap in capacities:
625 if cap.region in workload:
626 by_region[cap.region] = _capacity_entry_from(cap)
627 errors.extend(f"{cap.region}: {err}" for err in cap.telemetry_errors)
628 else:
629 for region in present:
630 try:
631 cap = checker.get_region_capacity(region)
632 except Exception as e:
633 errors.append(f"{region}: {e}")
634 else:
635 by_region[region] = _capacity_entry_from(cap)
636 errors.extend(f"{region}: {err}" for err in cap.telemetry_errors)
637 for region in workload:
638 if region in by_region:
639 continue
640 by_region[region] = _capacity_unavailable_entry()
641 if region not in present:
642 errors.append(f"{region}: regional stack {prefix}-{region} is absent or not readable")
643 elif not configured:
644 errors.append(
645 f"{region}: capacity telemetry needs the configured region list from cdk.json"
646 )
648 by_region = {region: by_region[region] for region in workload}
649 data = {"by_region": by_region}
650 statuses = [entry["telemetry_status"] for entry in by_region.values()]
651 incomplete = sum(1 for status in statuses if status != "complete")
653 if sweep_returned_empty and errors and all(s == STATUS_UNAVAILABLE for s in statuses):
654 return Section(
655 name=SECTION_CAPACITY,
656 status=STATUS_ERROR,
657 data=data,
658 reason="the capacity sweep failed for every region",
659 errors=errors,
660 )
661 if all(s == STATUS_UNAVAILABLE for s in statuses):
662 return Section(
663 name=SECTION_CAPACITY,
664 status=STATUS_UNAVAILABLE,
665 data=data,
666 reason="capacity telemetry could not be attempted in any workload region",
667 errors=errors,
668 )
669 if incomplete:
670 return Section(
671 name=SECTION_CAPACITY,
672 status=STATUS_PARTIAL,
673 data=data,
674 reason=f"{incomplete} of {len(statuses)} regions reported incomplete telemetry",
675 errors=errors,
676 )
677 return Section(name=SECTION_CAPACITY, status=STATUS_OK, data=data, errors=errors)
680# ---------------------------------------------------------------------------
681# inference
682# ---------------------------------------------------------------------------
684_ENDPOINT_FIELDS = ("endpoint_name", "desired_state", "target_regions", "namespace", "updated_at")
687def _gather_inference(config: GCOConfig) -> Section:
688 """Summarize inference endpoint desired state from the global registry."""
689 from cli.inference import get_inference_manager
691 manager = get_inference_manager(config)
692 try:
693 endpoints = manager.list_endpoints()
694 except ClientError as e:
695 if e.response.get("Error", {}).get("Code") != "ResourceNotFoundException":
696 raise
697 stack = config.global_stack_name
698 return Section(
699 name=SECTION_INFERENCE,
700 status=STATUS_UNAVAILABLE,
701 reason=(
702 f"the inference endpoint registry is not deployed; deploy the "
703 f"global stack with `gco stacks deploy {stack}`"
704 ),
705 errors=[str(e)],
706 )
708 listed = [
709 {field_name: endpoint.get(field_name) for field_name in _ENDPOINT_FIELDS}
710 for endpoint in endpoints
711 ]
712 totals = Counter(str(endpoint.get("desired_state")) for endpoint in endpoints)
713 # The registry read is a single unpaginated scan; the count makes any
714 # truncation attributable.
715 data = {"totals": dict(totals), "count": len(listed), "endpoints": listed}
716 status = STATUS_OK if listed else STATUS_EMPTY
717 return Section(name=SECTION_INFERENCE, status=status, data=data)
720def _gather_costs(config: GCOConfig, requested: bool) -> Section:
721 """Read the Cost Explorer summary and cost-allocation-tag status.
723 Tier 2: Cost Explorer bills per ``GetCostAndUsage`` request, so this
724 section is gathered only on explicit request. No by-region breakdown is
725 populated — it would need a second billed request.
726 """
727 if not requested:
728 return Section(
729 name=SECTION_COSTS,
730 status=STATUS_SKIPPED,
731 reason="not requested; pass --with-costs (Cost Explorer bills per request)",
732 )
734 from cli.costs import get_cost_tracker
736 tracker = get_cost_tracker(config)
737 summary = tracker.get_cost_summary(days=COST_WINDOW_DAYS)
739 errors: list[str] = []
740 tags: list[dict[str, str]] | None = None
741 try:
742 tags = [
743 {"tag_key": tag.get("tag_key", ""), "status": tag.get("status", "")}
744 for tag in tracker.get_cost_allocation_tag_status()
745 ]
746 except Exception as e:
747 errors.append(f"cost allocation tag status: {e}")
749 data: dict[str, Any] = {
750 "total": round(summary.total, 2),
751 "currency": summary.currency,
752 "window_days": COST_WINDOW_DAYS,
753 "period_start": summary.period_start,
754 "period_end": summary.period_end,
755 "by_service": [
756 {"service": item.service, "amount": round(item.amount, 2)}
757 for item in summary.by_service
758 ],
759 # Whether the tag filters this total depends on are active in
760 # Billing, so a near-zero total is not misread as near-zero spend.
761 "allocation_tags": tags,
762 "as_of": datetime.now(UTC).isoformat(),
763 }
764 if errors:
765 return Section(
766 name=SECTION_COSTS,
767 status=STATUS_PARTIAL,
768 data=data,
769 reason="cost total read, but the allocation-tag status could not be",
770 errors=errors,
771 )
772 status = STATUS_EMPTY if not data["by_service"] and data["total"] == 0 else STATUS_OK
773 return Section(name=SECTION_COSTS, status=status, data=data)
776_NODEPOOL_FIELDS = ("name", "status", "capacity_types", "instance_types")
779def _gather_nodepools(config: GCOConfig, requested: bool, workload: list[str]) -> Section:
780 """List Karpenter NodePools per region, probing reachability first.
782 Tier 3: the NodePool listing talks straight to the EKS API endpoint,
783 which is private by default; against a private endpoint it would block
784 until timeout. Each region's endpoint posture is probed first and a
785 non-public endpoint is reported ``unavailable`` — the Kubernetes call
786 is never attempted in that case.
787 """
788 if not requested:
789 return Section(
790 name=SECTION_NODEPOOLS,
791 status=STATUS_SKIPPED,
792 reason="not requested; pass --with-nodepools (requires cluster API reachability)",
793 )
794 if not workload:
795 return _regions_unavailable_section(SECTION_NODEPOOLS)
797 from cli import kubectl_helpers
798 from cli.nodepools import list_cluster_nodepools
800 by_region: dict[str, dict[str, Any]] = {}
801 errors: list[str] = []
802 listed = 0
803 private = 0
805 for region in workload:
806 cluster = f"{config.project_name}-{region}"
807 try:
808 access = kubectl_helpers.describe_cluster_access(cluster, region)
809 except Exception as e:
810 errors.append(f"{region}: {e}")
811 by_region[region] = {
812 "cluster": cluster,
813 "reachable": False,
814 "note": "cluster endpoint posture could not be determined",
815 }
816 continue
817 if not access.get("public"):
818 private += 1
819 by_region[region] = {
820 "cluster": cluster,
821 "reachable": False,
822 "note": (
823 f"cluster endpoint is private; open a tunnel with "
824 f"`gco cluster tunnel --region {region}`"
825 ),
826 }
827 continue
828 try:
829 pools = list_cluster_nodepools(cluster, region)
830 except Exception as e:
831 errors.append(f"{region}: {e}")
832 by_region[region] = {
833 "cluster": cluster,
834 "reachable": True,
835 "note": "nodepool listing failed",
836 }
837 continue
838 listed += 1
839 by_region[region] = {
840 "cluster": cluster,
841 "reachable": True,
842 "nodepools": [
843 {field_name: pool.get(field_name) for field_name in _NODEPOOL_FIELDS}
844 for pool in pools
845 ],
846 }
848 data = {"by_region": by_region}
849 if listed == len(workload):
850 return Section(name=SECTION_NODEPOOLS, status=STATUS_OK, data=data)
851 if listed:
852 return Section(
853 name=SECTION_NODEPOOLS,
854 status=STATUS_PARTIAL,
855 data=data,
856 reason=f"nodepools listed in {listed} of {len(workload)} regions",
857 errors=errors,
858 )
859 if private:
860 return Section(
861 name=SECTION_NODEPOOLS,
862 status=STATUS_UNAVAILABLE,
863 data=data,
864 reason=(
865 "no cluster endpoint is publicly reachable; open a tunnel with "
866 "`gco cluster tunnel` and use `gco nodepools list` through it"
867 ),
868 errors=errors,
869 )
870 return Section(
871 name=SECTION_NODEPOOLS,
872 status=STATUS_ERROR,
873 data=data,
874 reason="nodepools could not be listed in any region",
875 errors=errors,
876 )
879# ---------------------------------------------------------------------------
880# Findings
881# ---------------------------------------------------------------------------
884def _gather_policy(config: GCOConfig, requested: bool, workload: list[str]) -> Section:
885 """Compare the job-validation policy each region actually enforces.
887 Opt-in because it costs a CloudFormation describe plus an API call per
888 region, and because it needs the regional API bridge deployed.
890 Every region is deployed from the same ``cdk.json`` -- there are no
891 per-region policy overrides -- so a field that differs across regions means
892 at least one region is running a different deployment of that file. Nothing
893 else reports this: each region is individually healthy and self-consistent,
894 and the divergence only shows up as a manifest that is admitted in one
895 region and refused in another.
897 ``trusted_registries`` is compared with ECR hostnames stripped, because CDK
898 appends the project's own registries at synth time and those encode a region.
899 """
900 if not requested:
901 return Section(
902 name=SECTION_POLICY,
903 status=STATUS_SKIPPED,
904 reason="not requested; pass --with-policy (one API call per region)",
905 )
906 if not workload:
907 return _regions_unavailable_section(SECTION_POLICY)
909 from cli.aws_client import get_aws_client
910 from cli.job_policy import (
911 detect_policy_drift,
912 ecr_augmentation,
913 fetch_region_policies,
914 registry_drift,
915 )
917 policies = fetch_region_policies(get_aws_client(config), workload)
918 readable = [entry for entry in policies if entry.ok]
919 unreadable = {entry.region: entry.reason or "unknown" for entry in policies if not entry.ok}
921 drift = detect_policy_drift(policies)
922 registries = registry_drift(policies)
923 if registries is not None:
924 drift = [*drift, registries]
926 data: dict[str, Any] = {
927 "compared": [entry.region for entry in readable],
928 "unreadable": unreadable,
929 "agree": not drift,
930 "drift": [{"field": item.field, "values": item.values} for item in drift],
931 "ecr_augmentation": {r: h for r, h in ecr_augmentation(policies).items() if h},
932 "enforcement_gaps": {
933 entry.region: entry.enforcement_gaps for entry in readable if entry.enforcement_gaps
934 },
935 }
937 if not readable:
938 return Section(
939 name=SECTION_POLICY,
940 status=STATUS_UNAVAILABLE,
941 reason="no region's policy could be read",
942 data=data,
943 errors=[f"{region}: {reason}" for region, reason in sorted(unreadable.items())],
944 )
945 if unreadable:
946 return Section(
947 name=SECTION_POLICY,
948 status=STATUS_PARTIAL,
949 reason=f"{len(unreadable)} of {len(policies)} regions unreadable",
950 data=data,
951 errors=[f"{region}: {reason}" for region, reason in sorted(unreadable.items())],
952 )
953 if len(readable) < 2:
954 # One region cannot disagree with itself. Report the policy as read
955 # rather than implying agreement was verified.
956 return Section(name=SECTION_POLICY, status=STATUS_OK, data=data)
957 return Section(name=SECTION_POLICY, status=STATUS_OK, data=data)
960def derive_findings(sections: dict[str, Section]) -> list[Finding]:
961 """Derive the findings list from an already-gathered document.
963 A pure function over section data: it issues no AWS calls, so it cannot
964 fail in a new way or slow the gather down. The rule set is closed and
965 small on purpose — an open-ended heuristic layer becomes a source of
966 false alarms. Only expected stacks produce findings; optional stacks may
967 legitimately be undeployed. The result is ordered ``error`` before
968 ``warn``, each in document order.
969 """
970 errors: list[Finding] = []
971 warns: list[Finding] = []
973 stacks = sections.get(SECTION_STACKS)
974 if stacks is not None:
975 for entry in stacks.data.get("expected", []):
976 name = entry.get("name")
977 region = entry.get("region")
978 health = entry.get("health")
979 if health == HEALTH_UNHEALTHY:
980 errors.append(
981 Finding(
982 severity=SEVERITY_ERROR,
983 section=SECTION_STACKS,
984 message=f"{name} is {entry.get('status')} in {region}",
985 )
986 )
987 elif health == HEALTH_NOT_DEPLOYED:
988 # get_stack_status cannot distinguish a missing stack from
989 # denied access, so the wording must not assert absence.
990 warns.append(
991 Finding(
992 severity=SEVERITY_WARN,
993 section=SECTION_STACKS,
994 message=f"{name} is absent or not readable in {region}",
995 )
996 )
997 elif health == HEALTH_IN_PROGRESS:
998 warns.append(
999 Finding(
1000 severity=SEVERITY_WARN,
1001 section=SECTION_STACKS,
1002 message=f"{name} is {entry.get('status')} in {region}",
1003 )
1004 )
1006 queue = sections.get(SECTION_QUEUE)
1007 if queue is not None:
1008 for region, entry in queue.data.get("by_region", {}).items():
1009 depth = entry.get("dlq")
1010 if isinstance(depth, int) and depth > 0:
1011 plural = "" if depth == 1 else "s"
1012 warns.append(
1013 Finding(
1014 severity=SEVERITY_WARN,
1015 section=SECTION_QUEUE,
1016 message=f"{region} dead-letter queue holds {depth} message{plural}",
1017 )
1018 )
1020 capacity = sections.get(SECTION_CAPACITY)
1021 if capacity is not None:
1022 for region, entry in capacity.data.get("by_region", {}).items():
1023 telemetry = entry.get("telemetry_status")
1024 if telemetry == STATUS_UNAVAILABLE:
1025 errors.append(
1026 Finding(
1027 severity=SEVERITY_ERROR,
1028 section=SECTION_CAPACITY,
1029 message=f"{region} telemetry is unavailable",
1030 )
1031 )
1032 elif telemetry == STATUS_PARTIAL:
1033 signals = ", ".join(entry.get("unavailable_signals", [])) or "unknown"
1034 warns.append(
1035 Finding(
1036 severity=SEVERITY_WARN,
1037 section=SECTION_CAPACITY,
1038 message=f"{region} telemetry is partial (unavailable: {signals})",
1039 )
1040 )
1042 jobs = sections.get(SECTION_JOBS)
1043 if jobs is not None and jobs.data and jobs.data.get("complete", True) is False:
1044 evaluated = jobs.data.get("records_evaluated")
1045 warns.append(
1046 Finding(
1047 severity=SEVERITY_WARN,
1048 section=SECTION_JOBS,
1049 message=(
1050 f"the job-count scan was truncated after {evaluated} records; "
1051 "totals are a floor, not a count"
1052 ),
1053 )
1054 )
1056 policy = sections.get(SECTION_POLICY)
1057 if policy is not None and policy.status in {STATUS_OK, STATUS_PARTIAL}:
1058 for item in policy.data.get("drift", []):
1059 field_name = item.get("field")
1060 values = item.get("values", {})
1061 spread = "; ".join(f"{region}={value}" for region, value in sorted(values.items()))
1062 warns.append(
1063 Finding(
1064 severity=SEVERITY_WARN,
1065 section=SECTION_POLICY,
1066 message=(
1067 f"{field_name} differs across regions ({spread}) — there are no "
1068 f"per-region policy overrides, so a region is running a different "
1069 f"deployment of cdk.json"
1070 ),
1071 )
1072 )
1073 for region, namespaces in sorted(policy.data.get("enforcement_gaps", {}).items()):
1074 warns.append(
1075 Finding(
1076 severity=SEVERITY_WARN,
1077 section=SECTION_POLICY,
1078 message=(
1079 f"{region} cannot read the live ResourceQuota/LimitRange for "
1080 f"{', '.join(namespaces)}, so only its front-door caps are "
1081 f"reportable (check the manifest-processor Role)"
1082 ),
1083 )
1084 )
1086 return errors + warns
1089# ---------------------------------------------------------------------------
1090# Verdict derivation
1091# ---------------------------------------------------------------------------
1094def _derive_overall(sections: dict[str, Section], findings: list[Finding]) -> tuple[str, list[str]]:
1095 """Derive the ``overall`` verdict and the responsible-section list.
1097 The document is degraded when any section is partial, unavailable, or in
1098 error, or when any finding is present. The ``degraded`` list names the
1099 sections with a degraded status or an error-severity finding, in section
1100 order; a warn finding flips the verdict without adding its section.
1101 """
1102 responsible = {
1103 name for name, section in sections.items() if section.status in _DEGRADED_STATUSES
1104 }
1105 responsible.update(
1106 finding.section for finding in findings if finding.severity == SEVERITY_ERROR
1107 )
1108 if responsible or findings:
1109 return OVERALL_DEGRADED, [name for name in SECTION_ORDER if name in responsible]
1110 return OVERALL_OK, []
1113# ---------------------------------------------------------------------------
1114# Orchestrator
1115# ---------------------------------------------------------------------------
1118def gather_fleet_status(
1119 config: GCOConfig,
1120 *,
1121 region: str | None = None,
1122 with_costs: bool = False,
1123 with_nodepools: bool = False,
1124 with_policy: bool = False,
1125 costs_cache: Section | None = None,
1126) -> FleetStatus:
1127 """Gather every section and assemble the fleet status document.
1129 Always returns a document: sections that cannot be gathered degrade
1130 individually and the rest are unaffected.
1132 ``costs_cache`` reuses a previously gathered ``costs`` section instead
1133 of issuing a new Cost Explorer request. It exists for watch mode's
1134 in-process rate limit only; nothing is ever written to disk.
1135 """
1136 regions_section = _run_section(SECTION_REGIONS, lambda: resolve_regions(config, region))
1137 workload = _workload_regions(regions_section)
1138 configured = [item for item in _load_cdk_json().get("regional", []) if isinstance(item, str)]
1140 # One direct describe per regional stack, shared by the queue and
1141 # capacity gates (and accepted as a duplicate of the stacks section's
1142 # own read). An empty result keeps the discovery-based managers — whose
1143 # empty-discovery fallback scans every AWS region — from being invoked.
1144 regional_probe: dict[str, StackInfo | None] = {}
1145 if workload:
1146 try:
1147 regional_probe = _probe_regional_stacks(config, workload)
1148 except Exception as e:
1149 logger.debug("Regional stack probe failed: %s", e)
1150 regional_probe = dict.fromkeys(workload)
1152 gatherers: dict[str, Callable[[], Section]] = {}
1153 if workload:
1154 gatherers[SECTION_STACKS] = lambda: _gather_stacks(config, regions_section.data, workload)
1155 gatherers[SECTION_QUEUE] = lambda: _gather_queue(
1156 config, workload, regional_probe, bool(configured)
1157 )
1158 gatherers[SECTION_CAPACITY] = lambda: _gather_capacity(
1159 config, workload, configured, regional_probe
1160 )
1161 gatherers[SECTION_JOBS] = lambda: _gather_jobs(config, region)
1162 gatherers[SECTION_INFERENCE] = lambda: _gather_inference(config)
1163 if with_costs and costs_cache is not None:
1164 reused_costs = costs_cache
1165 gatherers[SECTION_COSTS] = lambda: reused_costs
1166 else:
1167 gatherers[SECTION_COSTS] = lambda: _gather_costs(config, with_costs)
1168 gatherers[SECTION_NODEPOOLS] = lambda: _gather_nodepools(config, with_nodepools, workload)
1169 gatherers[SECTION_POLICY] = lambda: _gather_policy(config, with_policy, workload)
1171 sections: dict[str, Section] = {SECTION_REGIONS: regions_section}
1172 if not workload:
1173 for name in sorted(_PER_REGION_SECTIONS):
1174 sections[name] = _regions_unavailable_section(name)
1175 sections.update(_run_sections_concurrently(gatherers))
1177 findings = derive_findings(sections)
1178 overall, degraded = _derive_overall(sections, findings)
1180 return FleetStatus(
1181 generated_at=datetime.now(UTC).isoformat(),
1182 project_name=config.project_name,
1183 overall=overall,
1184 degraded=degraded,
1185 findings=findings,
1186 sections={name: sections[name] for name in SECTION_ORDER},
1187 )