Coverage for cli / eks_capabilities.py: 100.00%
101 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Configured-vs-live view of the EKS Capabilities on a GCO regional cluster.
3Backs ``gco stacks capabilities status`` and the ``eks_capabilities_status`` MCP
4tool. EKS Capabilities are AWS-managed ACK and kro installations that GCO
5attaches per cluster from the opt-in ``cdk.json`` ``eks_capabilities`` block
6(see ``docs/EKS_CAPABILITIES.md``). This module answers the operator's
7question in one document: what does cdk.json say should be attached here, what
8is actually attached (``ListCapabilities`` / ``DescribeCapability``), is each
9one ``ACTIVE``, and where do the two disagree.
11Deliberately AWS-API only: no kubectl, so it works from any host with AWS
12credentials, including against a private-endpoint cluster.
14Pure functions (``build_status``) are separated from the AWS calls
15(``describe_live_capabilities``) so the merge logic is unit-testable without a
16fake EKS endpoint. Imports nothing from ``aws_cdk``: the CLI runs without the
17CDK toolchain installed.
18"""
20from __future__ import annotations
22import json
23from collections.abc import Iterable, Mapping
24from datetime import datetime
25from pathlib import Path
26from typing import Any
28from gco.eks_capabilities_config import (
29 CAPABILITY_TYPE_API_NAMES,
30 EKS_CAPABILITIES_CONTEXT_KEY,
31 EKS_CAPABILITY_TYPES,
32 capability_enabled_in_region,
33 merge_eks_capabilities_overrides,
34 parse_eks_capabilities_overrides,
35 validate_eks_capabilities_config,
36)
38#: The only capability status that means "attached and serving".
39ACTIVE_STATUS = "ACTIVE"
41#: ``AWS::EKS::Capability`` ``Type`` value -> cdk.json spelling.
42_API_TYPE_TO_CONFIG: dict[str, str] = {
43 api_name: type_name for type_name, api_name in CAPABILITY_TYPE_API_NAMES.items()
44}
47def capability_name(project_name: str, type_name: str) -> str:
48 """The deterministic name the regional stack gives each capability."""
49 return f"{project_name}-{type_name}"
52def cluster_name_for(project_name: str, region: str) -> str:
53 """The regional cluster's name (``<project>-<region>``)."""
54 return f"{project_name}-{region}"
57def load_eks_capabilities_config(
58 cdk_json_path: Path | None = None,
59 *,
60 overrides: object = None,
61) -> dict[str, Any]:
62 """The normalized ``eks_capabilities`` block from cdk.json.
64 Validated with the same rules the CDK app applies (including the per-type
65 ``regions`` subsets against ``deployment_regions.regional``), so a block
66 the next deploy would reject is reported here instead of being rendered as
67 if it were live intent. ``overrides`` is the run-scoped
68 ``eks_capabilities_overrides`` value (a JSON object string or mapping) a
69 harness deployed with, merged in exactly as the CDK app merges it. Raises
70 ``RuntimeError`` when cdk.json is missing and
71 ``gco.eks_capabilities_config.EksCapabilitiesConfigError`` (a
72 ``ValueError``) when the block is malformed.
73 """
74 path = cdk_json_path or Path.cwd() / "cdk.json"
75 if not path.exists():
76 raise RuntimeError(f"cdk.json not found at {path}")
77 with open(path, encoding="utf-8") as handle:
78 document = json.load(handle)
79 context = document.get("context") if isinstance(document, dict) else None
80 if not isinstance(context, dict):
81 context = {}
82 deployment_regions = context.get("deployment_regions")
83 regional = deployment_regions.get("regional") if isinstance(deployment_regions, dict) else None
84 raw = merge_eks_capabilities_overrides(
85 context.get(EKS_CAPABILITIES_CONTEXT_KEY),
86 parse_eks_capabilities_overrides(overrides),
87 )
88 return validate_eks_capabilities_config(raw, regional if isinstance(regional, list) else None)
91def describe_live_capabilities(eks_client: Any, cluster_name: str) -> list[dict[str, Any]]:
92 """Every capability attached to ``cluster_name``, fully described.
94 ``ListCapabilities`` returns summaries (name, type, status, version);
95 ``DescribeCapability`` adds the role ARN and health issues, so each
96 summary is described in turn. Raises whatever boto3
97 raises (the caller decides how a missing cluster reads).
98 """
99 summaries: list[dict[str, Any]] = []
100 paginator = eks_client.get_paginator("list_capabilities")
101 for page in paginator.paginate(clusterName=cluster_name):
102 summaries.extend(item for item in page.get("capabilities") or [] if isinstance(item, dict))
103 described: list[dict[str, Any]] = []
104 for summary in summaries:
105 name = summary.get("capabilityName")
106 if not isinstance(name, str) or not name:
107 continue
108 response = eks_client.describe_capability(clusterName=cluster_name, capabilityName=name)
109 detail = response.get("capability") if isinstance(response, dict) else None
110 described.append(detail if isinstance(detail, dict) else dict(summary))
111 return described
114def _iso(value: Any) -> str | None:
115 if isinstance(value, datetime):
116 return value.isoformat()
117 return str(value) if value else None
120def _health_issues(detail: Mapping[str, Any]) -> list[str]:
121 health = detail.get("health")
122 issues = health.get("issues") if isinstance(health, Mapping) else None
123 rendered: list[str] = []
124 for issue in issues or []:
125 if not isinstance(issue, Mapping):
126 continue
127 code = str(issue.get("code") or "Issue")
128 message = str(issue.get("message") or "").strip()
129 rendered.append(f"{code}: {message}" if message else code)
130 return rendered
133def _drift(*, configured: bool, detail: Mapping[str, Any] | None, stack_name: str) -> str | None:
134 """One sentence naming how cdk.json and the cluster disagree, or ``None``."""
135 if configured and detail is None:
136 return f"configured in cdk.json but not attached; run 'gco stacks deploy {stack_name} -y'"
137 if detail is not None and not configured:
138 return (
139 "attached but disabled in cdk.json; the next 'gco stacks deploy "
140 f"{stack_name}' removes it (RETAIN keeps what it installed)"
141 )
142 if detail is not None:
143 status = str(detail.get("status") or "UNKNOWN")
144 if status != ACTIVE_STATUS:
145 issues = _health_issues(detail)
146 suffix = f" ({'; '.join(issues)})" if issues else ""
147 return f"status is {status}, expected {ACTIVE_STATUS}{suffix}"
148 return None
151def build_status(
152 *,
153 region: str,
154 project_name: str,
155 config: Mapping[str, Any],
156 live: Iterable[Mapping[str, Any]],
157 cluster_found: bool = True,
158) -> dict[str, Any]:
159 """Merge the cdk.json intent with the described capabilities for one region.
161 One row per capability type GCO knows (in canonical order), each stating
162 whether cdk.json enables it for this region, whether it is attached, its
163 live status/version/ARNs, health issues, and a ``drift`` sentence when the
164 two disagree. Capabilities on the cluster that do not carry GCO's
165 ``<project>-<type>`` name are listed under ``unmanaged`` so an operator's
166 hand-made capability is visible but never mistaken for GCO's. ``healthy``
167 is true only when nothing drifts.
168 """
169 cluster = cluster_name_for(project_name, region)
170 stack_name = cluster
171 live_by_name: dict[str, Mapping[str, Any]] = {}
172 for item in live:
173 name = item.get("capabilityName")
174 if isinstance(name, str):
175 live_by_name[name] = item
177 rows: list[dict[str, Any]] = []
178 for type_name in EKS_CAPABILITY_TYPES:
179 name = capability_name(project_name, type_name)
180 configured = capability_enabled_in_region(config, type_name, region)
181 detail = live_by_name.pop(name, None)
182 row: dict[str, Any] = {
183 "type": type_name,
184 "capability_name": name,
185 "configured": configured,
186 "deployed": detail is not None,
187 "status": str(detail["status"]) if detail and detail.get("status") else None,
188 "version": str(detail["version"]) if detail and detail.get("version") else None,
189 "arn": str(detail["arn"]) if detail and detail.get("arn") else None,
190 "role_arn": str(detail["roleArn"]) if detail and detail.get("roleArn") else None,
191 "health_issues": _health_issues(detail) if detail else [],
192 "modified_at": _iso(detail.get("modifiedAt")) if detail else None,
193 "drift": _drift(configured=configured, detail=detail, stack_name=stack_name),
194 }
195 rows.append(row)
197 unmanaged = [
198 {
199 "capability_name": name,
200 "type": _API_TYPE_TO_CONFIG.get(str(item.get("type")), str(item.get("type"))),
201 "status": str(item.get("status")) if item.get("status") else None,
202 "arn": str(item.get("arn")) if item.get("arn") else None,
203 }
204 for name, item in sorted(live_by_name.items())
205 ]
207 return {
208 "region": region,
209 "cluster_name": cluster,
210 "cluster_found": cluster_found,
211 "capabilities": rows,
212 "unmanaged": unmanaged,
213 "healthy": cluster_found and all(row["drift"] is None for row in rows),
214 }
217def capabilities_status(
218 region: str,
219 project_name: str,
220 *,
221 config: Mapping[str, Any] | None = None,
222 eks_client: Any | None = None,
223) -> dict[str, Any]:
224 """``build_status`` for one region against the real EKS API.
226 A cluster that does not exist (never deployed, or destroyed) is reported
227 with ``cluster_found: false`` and no live rows rather than as an error, so
228 ``--all-regions`` keeps going and a configured-but-undeployed region reads
229 as drift instead of a crash.
230 """
231 from botocore.exceptions import ClientError
233 if config is None:
234 config = load_eks_capabilities_config()
235 cluster = cluster_name_for(project_name, region)
236 if eks_client is None:
237 import boto3
239 eks_client = boto3.client("eks", region_name=region)
240 try:
241 live = describe_live_capabilities(eks_client, cluster)
242 cluster_found = True
243 except ClientError as exc:
244 if exc.response.get("Error", {}).get("Code") != "ResourceNotFoundException":
245 raise
246 live = []
247 cluster_found = False
248 return build_status(
249 region=region,
250 project_name=project_name,
251 config=config,
252 live=live,
253 cluster_found=cluster_found,
254 )