Coverage for cli / commands / cluster_cmd.py: 100.00%
104 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""GCO cluster connectivity commands.
3``gco cluster tunnel`` opens an AWS SSM Session Manager tunnel to a cluster's
4PRIVATE EKS API endpoint so ``kubectl`` works from a laptop outside the VPC —
5optionally provisioning a self-terminating ephemeral bastion (``--via-ssm
6auto``). It reuses the same tunnel/bastion machinery as ``gco monitoring open``
7(:mod:`cli.cluster_tunnel`).
9``--print`` emits the ready-to-run tunnel + ``kubectl`` commands (a connection
10plan) instead of opening the tunnel — the request/response form the MCP
11``cluster_tunnel_command`` tool returns, and handy for scripting or when you
12already have your own SSM instance.
13"""
15from __future__ import annotations
17import socket
18import sys
19import time
20from typing import Any
22import click
24from ..cluster_tunnel import (
25 DEFAULT_API_LOCAL_PORT,
26 ROUTE_DIRECT,
27 open_api_server_tunnel,
28 require_private_access,
29 resolve_region,
30 resolve_tunnel_plan,
31)
32from ..config import GCOConfig
33from ..ephemeral_bastion import DEFAULT_TTL_MINUTES
34from ..output import get_output_formatter
36pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
39@click.group()
40@pass_config
41def cluster(config: Any) -> None:
42 """Cluster connectivity helpers (SSM tunnel to the private EKS API)."""
45@cluster.command("tunnel")
46@click.option("--region", help="Cluster region (defaults to the first cdk.json regional entry).")
47@click.option(
48 "--via-ssm",
49 "via_ssm",
50 metavar="INSTANCE_ID|auto",
51 help=(
52 "Tunnel through an SSM-managed instance id, or 'auto' to provision a "
53 "self-terminating ephemeral bastion and tear it down on exit. Used whenever "
54 "private endpoint access is on, PUBLIC_AND_PRIVATE included."
55 ),
56)
57@click.option(
58 "--local-port",
59 type=int,
60 default=DEFAULT_API_LOCAL_PORT,
61 show_default=True,
62 help="Local port to bind for the API tunnel.",
63)
64@click.option(
65 "--print",
66 "print_plan",
67 is_flag=True,
68 help="Print the tunnel + kubectl commands (connection plan) instead of opening the tunnel.",
69)
70@click.option(
71 "--bastion-ttl-minutes",
72 type=int,
73 default=DEFAULT_TTL_MINUTES,
74 show_default=True,
75 help="Self-terminate backstop (minutes) for an `--via-ssm auto` bastion.",
76)
77@click.option(
78 "--yes",
79 "-y",
80 "assume_yes",
81 is_flag=True,
82 help="Skip the confirmation prompt when provisioning an `--via-ssm auto` bastion.",
83)
84@pass_config
85def tunnel_cmd(
86 config: Any,
87 region: str | None,
88 via_ssm: str | None,
89 local_port: int,
90 print_plan: bool,
91 bastion_ttl_minutes: int,
92 assume_yes: bool,
93) -> None:
94 """Open (or ``--print``) an SSM tunnel to a cluster's private EKS API endpoint.
96 Interactive mode holds the tunnel open in the foreground (Ctrl-C to stop) and
97 prints the ``kubectl`` flags to use in another shell. Pass ``--via-ssm
98 <instance-id>`` to tunnel through an existing SSM-managed instance, or
99 ``--via-ssm auto`` to provision a minimal, self-terminating ephemeral bastion
100 for the session. Either is honored whenever private endpoint access is on,
101 including a PUBLIC_AND_PRIVATE cluster whose public endpoint this network
102 cannot reach. Without ``--via-ssm``, a public endpoint is used directly.
103 """
104 formatter = get_output_formatter(config)
105 target_region = resolve_region(config, region)
106 cluster_name = f"{config.project_name}-{target_region}"
108 if print_plan:
109 _print_connection_plan(config, formatter, cluster_name, target_region, via_ssm, local_port)
110 return
112 from ..kubectl_helpers import update_kubeconfig
114 try:
115 update_kubeconfig(cluster_name, target_region)
116 except (RuntimeError, ValueError) as exc:
117 formatter.print_error(str(exc))
118 sys.exit(1)
120 try:
121 with open_api_server_tunnel(
122 formatter,
123 cluster=cluster_name,
124 region=target_region,
125 via_ssm=via_ssm,
126 local_port=local_port,
127 bastion_ttl_minutes=bastion_ttl_minutes,
128 assume_yes=assume_yes,
129 ) as session:
130 if session.active:
131 formatter.print_success(
132 f"SSM tunnel open: {session.server} → {session.plan.endpoint} (Ctrl-C to stop)"
133 )
134 kubectl_command = "kubectl " + " ".join(session.plan.kubectl_flags()) + " get nodes"
135 formatter.print_info(_kubectl_hint(kubectl_command))
136 _block_until_interrupt()
137 elif session.route == ROUTE_DIRECT:
138 formatter.print_success(
139 f"{cluster_name} has a PUBLIC API endpoint — kubectl reaches it directly "
140 "(after `aws eks update-kubeconfig`); no tunnel needed."
141 )
142 # A private endpoint with no --via-ssm, or one whose access mode could
143 # not be read, was already explained by the context manager; there is
144 # nothing to hold open.
145 except (RuntimeError, ValueError) as exc:
146 formatter.print_error(str(exc))
147 sys.exit(1)
150def _print_connection_plan(
151 config: Any,
152 formatter: Any,
153 cluster_name: str,
154 region: str,
155 via_ssm: str | None,
156 local_port: int,
157) -> None:
158 """Resolve and emit the connection plan (JSON under -o json, else commands).
160 The plan follows the interactive route rule: ``--via-ssm`` selects the SSM
161 tunnel whenever private access is on, and is rejected when it is off.
162 """
163 try:
164 plan = resolve_tunnel_plan(cluster_name, region, local_port=local_port)
165 if via_ssm:
166 require_private_access(plan, via_ssm)
167 payload = plan.as_dict(via_ssm)
168 except (RuntimeError, ValueError) as exc:
169 formatter.print_error(f"Failed to resolve tunnel plan: {exc}")
170 sys.exit(1)
172 if config.output_format in ("json", "yaml"):
173 formatter.print(payload)
174 else:
175 _echo_plan_human(payload)
178def _echo_plan_human(payload: dict[str, Any]) -> None:
179 """Print the connection plan as copy-paste shell commands."""
180 header = f"# {payload['cluster']} ({payload['region']})"
181 mode = payload["access_mode"]
182 lines: list[str] = []
183 if payload.get("reachable") == "direct":
184 lines.append(f"{header}: {mode} endpoint — kubectl reaches the public endpoint directly.")
185 lines.append("# " + payload["note"])
186 lines.append(" ".join(payload["update_kubeconfig"]))
187 else:
188 lines.append(f"{header}: {mode} endpoint — reach the private endpoint over an SSM tunnel.")
189 lines.append("# 1) Ensure a kubeconfig context exists:")
190 lines.append(" ".join(payload["update_kubeconfig"]))
191 lines.append("# 2) Open the tunnel in one shell (keep it running):")
192 lines.append(payload.get("ssm_command_str") or payload.get("ssm_command_template", ""))
193 lines.append("# 3) In another shell, run kubectl through the tunnel:")
194 lines.append(payload["kubectl_example"])
195 if payload.get("note"):
196 lines.append("# " + payload["note"])
197 click.echo("\n".join(lines))
200def _kubectl_hint(kubectl_command: str) -> str:
201 """Say where to run kubectl so it actually reaches the tunnel.
203 The tunnel listens on 127.0.0.1 of the process that opened it. Inside the
204 ``gco-dev`` container (the ``gco`` shell function from
205 ``scripts/setup-dev-alias.sh``) that is the container's own loopback, which
206 neither the host nor a second ``gco`` invocation (a fresh container) can
207 reach, so kubectl has to run in this same container.
208 """
209 from ..stacks import _running_in_dev_container
211 if not _running_in_dev_container():
212 return f"Run kubectl in another shell with:\n {kubectl_command}"
213 container = socket.gethostname()
214 return (
215 "This tunnel listens inside the gco-dev container, so kubectl on the host or in "
216 "another `gco` call cannot reach it. From another shell, run kubectl in this "
217 "container (use finch or podman instead of docker if that is your runtime):\n"
218 f" docker exec -it {container} {kubectl_command}"
219 )
222def _block_until_interrupt() -> None: # pragma: no cover - interactive wait
223 """Hold the process (and thus the tunnel) open until Ctrl-C."""
224 try:
225 while True:
226 time.sleep(3600)
227 except KeyboardInterrupt:
228 pass
231_DOCTOR_STATUS_MARKS = {"ok": "✓", "warn": "⚠", "fail": "✗", "unknown": "?"}
234@cluster.command("doctor")
235@click.option("--region", help="Cluster region (defaults to the first cdk.json regional entry).")
236@pass_config
237def doctor_cmd(config: Any, region: str | None) -> None:
238 """Diagnose EKS API access: reachability, authentication, authorization.
240 The three layers fail with misleading and overlapping symptoms — kubectl
241 timeouts (endpoint mode or CIDR allowlist), 'Unauthorized' (no EKS access
242 entry), RBAC 'Forbidden' (no associated access policy) — and a stale
243 kubeconfig context for a destroyed cluster produces the same 'no such
244 host' as a private endpoint. Doctor reports each layer separately with
245 the remedy for exactly what it found, and exits nonzero if any layer
246 fails.
247 """
248 from .. import cluster_doctor
250 formatter = get_output_formatter(config)
251 target_region = resolve_region(config, region)
252 cluster_name = f"{config.project_name}-{target_region}"
254 probe = cluster_doctor.probe_cluster(cluster_name, target_region)
255 checks = cluster_doctor.diagnose(probe)
257 if config.output_format in ("json", "yaml"):
258 formatter.print(
259 {
260 "cluster": cluster_name,
261 "region": target_region,
262 "checks": [check.as_dict() for check in checks],
263 }
264 )
265 else:
266 click.echo(f"# {cluster_name} ({target_region})")
267 for check in checks:
268 mark = _DOCTOR_STATUS_MARKS.get(check.status, "?")
269 click.echo(f"{mark} [{check.layer}] {check.finding}")
270 if check.remedy:
271 click.echo(f" remedy: {check.remedy}")
273 if any(check.status == "fail" for check in checks):
274 sys.exit(1)