Coverage for cli/commands/cluster_cmd.py: 98.82%

1"""GCO cluster connectivity commands. 

2 

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`). 

8 

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""" 

14 

15from __future__ import annotations 

16 

17import sys 

18import time 

19from typing import Any 

20 

21import click 

22 

23from ..cluster_tunnel import ( 

24 AUTO_BASTION, 

25 DEFAULT_API_LOCAL_PORT, 

26 open_api_server_tunnel, 

27 resolve_region, 

28 resolve_tunnel_plan, 

29) 

30from ..config import GCOConfig 

31from ..ephemeral_bastion import DEFAULT_TTL_MINUTES 

32from ..output import get_output_formatter 

33 

34pass_config = click.make_pass_decorator(GCOConfig, ensure=True) 

35 

36 

37@click.group() 

38@pass_config 

39def cluster(config: Any) -> None: 

40 """Cluster connectivity helpers (SSM tunnel to the private EKS API).""" 

41 

42 

43@cluster.command("tunnel") 

44@click.option("--region", help="Cluster region (defaults to the first cdk.json regional entry).") 

45@click.option( 

46 "--via-ssm", 

47 "via_ssm", 

48 metavar="INSTANCE_ID|auto", 

49 help=( 

50 "Tunnel through an SSM-managed instance id, or 'auto' to provision a " 

51 "self-terminating ephemeral bastion and tear it down on exit." 

52 ), 

53) 

54@click.option( 

55 "--local-port", 

56 type=int, 

57 default=DEFAULT_API_LOCAL_PORT, 

58 show_default=True, 

59 help="Local port to bind for the API tunnel.", 

60) 

61@click.option( 

62 "--print", 

63 "print_plan", 

64 is_flag=True, 

65 help="Print the tunnel + kubectl commands (connection plan) instead of opening the tunnel.", 

66) 

67@click.option( 

68 "--bastion-ttl-minutes", 

69 type=int, 

70 default=DEFAULT_TTL_MINUTES, 

71 show_default=True, 

72 help="Self-terminate backstop (minutes) for an `--via-ssm auto` bastion.", 

73) 

74@click.option( 

75 "--yes", 

76 "-y", 

77 "assume_yes", 

78 is_flag=True, 

79 help="Skip the confirmation prompt when provisioning an `--via-ssm auto` bastion.", 

80) 

81@pass_config 

82def tunnel_cmd( 

83 config: Any, 

84 region: str | None, 

85 via_ssm: str | None, 

86 local_port: int, 

87 print_plan: bool, 

88 bastion_ttl_minutes: int, 

89 assume_yes: bool, 

90) -> None: 

91 """Open (or ``--print``) an SSM tunnel to a cluster's private EKS API endpoint. 

92 

93 Interactive mode holds the tunnel open in the foreground (Ctrl-C to stop) and 

94 prints the ``kubectl`` flags to use in another shell. On a private-endpoint 

95 cluster pass ``--via-ssm <instance-id>`` to tunnel through an existing 

96 SSM-managed instance, or ``--via-ssm auto`` to provision a minimal, 

97 self-terminating ephemeral bastion for the session. 

98 """ 

99 formatter = get_output_formatter(config) 

100 target_region = resolve_region(config, region) 

101 cluster_name = f"{config.project_name}-{target_region}" 

102 

103 if print_plan: 

104 _print_connection_plan(config, formatter, cluster_name, target_region, via_ssm, local_port) 

105 return 

106 

107 from ..kubectl_helpers import update_kubeconfig 

108 

109 try: 

110 update_kubeconfig(cluster_name, target_region) 

111 except (RuntimeError, ValueError) as exc: 

112 formatter.print_error(str(exc)) 

113 sys.exit(1) 

114 

115 try: 

116 with open_api_server_tunnel( 

117 formatter, 

118 cluster=cluster_name, 

119 region=target_region, 

120 via_ssm=via_ssm, 

121 local_port=local_port, 

122 bastion_ttl_minutes=bastion_ttl_minutes, 

123 assume_yes=assume_yes, 

124 ) as session: 

125 if session.active: 

126 formatter.print_success( 

127 f"SSM tunnel open: {session.server} → {session.plan.endpoint} (Ctrl-C to stop)" 

128 ) 

129 formatter.print_info( 

130 "Run kubectl in another shell with:\n kubectl " 

131 + " ".join(session.plan.kubectl_flags()) 

132 + " get nodes" 

133 ) 

134 _block_until_interrupt() 

135 elif session.plan.public: 

136 formatter.print_success( 

137 f"{cluster_name} has a PUBLIC API endpoint — kubectl reaches it directly " 

138 "(after `aws eks update-kubeconfig`); no tunnel needed." 

139 ) 

140 # A private endpoint with no --via-ssm was already explained by the 

141 # context manager (private_endpoint_guidance); nothing to hold open. 

142 except (RuntimeError, ValueError) as exc: 

143 formatter.print_error(str(exc)) 

144 sys.exit(1) 

145 

146 

147def _print_connection_plan( 

148 config: Any, 

149 formatter: Any, 

150 cluster_name: str, 

151 region: str, 

152 via_ssm: str | None, 

153 local_port: int, 

154) -> None: 

155 """Resolve and emit the connection plan (JSON under -o json, else commands).""" 

156 instance_id = via_ssm if (via_ssm and via_ssm != AUTO_BASTION) else None 

157 try: 

158 plan = resolve_tunnel_plan(cluster_name, region, local_port=local_port) 

159 payload = plan.as_dict(instance_id) 

160 except (RuntimeError, ValueError) as exc: 

161 formatter.print_error(f"Failed to resolve tunnel plan: {exc}") 

162 sys.exit(1) 

163 

164 if config.output_format in ("json", "yaml"): 

165 formatter.print(payload) 

166 else: 

167 _echo_plan_human(payload) 

168 

169 

170def _echo_plan_human(payload: dict[str, Any]) -> None: 

171 """Print the connection plan as copy-paste shell commands.""" 

172 header = f"# {payload['cluster']} ({payload['region']})" 

173 lines: list[str] = [] 

174 if payload.get("reachable") == "direct": 

175 lines.append(f"{header}: PUBLIC endpoint — kubectl reaches it directly.") 

176 lines.append("# " + payload["note"]) 

177 lines.append(" ".join(payload["update_kubeconfig"])) 

178 else: 

179 lines.append(f"{header}: PRIVATE endpoint — reach it over an SSM tunnel.") 

180 lines.append("# 1) Ensure a kubeconfig context exists:") 

181 lines.append(" ".join(payload["update_kubeconfig"])) 

182 lines.append("# 2) Open the tunnel in one shell (keep it running):") 

183 lines.append(payload.get("ssm_command_str") or payload.get("ssm_command_template", "")) 

184 lines.append("# 3) In another shell, run kubectl through the tunnel:") 

185 lines.append(payload["kubectl_example"]) 

186 if payload.get("note"): 186 ↛ 188line 186 didn't jump to line 188 because the condition on line 186 was always true

187 lines.append("# " + payload["note"]) 

188 click.echo("\n".join(lines)) 

189 

190 

191def _block_until_interrupt() -> None: # pragma: no cover - interactive wait 

192 """Hold the process (and thus the tunnel) open until Ctrl-C.""" 

193 try: 

194 while True: 

195 time.sleep(3600) 

196 except KeyboardInterrupt: 

197 pass