Coverage for scripts / example_job_validation / kube.py: 100.00%
313 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Cluster access for live harnesses: SSM tunnel plus kubectl execution.
3Both live harnesses pass an isolated kubeconfig in their private report
4directory. AWS CLI and kubectl receive ``--kubeconfig``, and every nested
5``gco`` process receives ``KUBECONFIG`` (:func:`session_environment` hands
6drivers the same environment), so a run never reads or rewrites
7``~/.kube/config`` or a ``KUBECONFIG`` the operator's shell exports. Without
8an explicit path the default kubeconfig is used.
10A session that reaches the API through an SSM tunnel also keeps the tunnel
11carrying traffic. A live run lost seven examples to a Session Manager session
12that stalled for over an hour: the local plugin kept accepting connections,
13but no TLS handshake completed, so every kubectl call failed with ``TLS
14handshake timeout`` and the watchers read the submitted Jobs as missing. A
15watchdog now probes a TLS handshake through the tunnel every 30 seconds and,
16after two failures in a row (or at once when the session exited), reopens the
17session on the same local port through the same instance, so the kubeconfig
18and every ``gco`` process that reads it keep working. A call that fails on
19the tunnel's transport reopens a broken tunnel at once and is repeated one
20time (:meth:`SessionKubectl.through_tunnel`).
21"""
23from __future__ import annotations
25import os
26import socket
27import ssl
28import stat
29import subprocess
30import threading
31import time
32from collections.abc import Callable, Iterator, Mapping
33from contextlib import contextmanager
34from datetime import UTC, datetime
35from pathlib import Path
36from typing import Any
38import yaml
40#: How the harness invokes kubectl; a function taking kubectl args and
41#: returning (exit_code, stdout, stderr).
42KubectlRunner = Callable[..., tuple[int, str, str]]
44#: How often the watchdog probes the tunnel, how long one TLS handshake may
45#: take (client-go's own handshake timeout), and how many failed probes in a
46#: row it takes before the watchdog reopens a session that is still running.
47_TUNNEL_CHECK_INTERVAL_SECONDS = 30.0
48_TUNNEL_HANDSHAKE_TIMEOUT_SECONDS = 10.0
49_TUNNEL_FAILURES_BEFORE_REOPEN = 2
50#: A reopen that failed (the instance is gone, SSM refused the session) is
51#: not tried again sooner than this, so parallel callers cannot stampede SSM.
52_TUNNEL_REOPEN_BACKOFF_SECONDS = 60.0
53_TUNNEL_EVENT_TEXT_LIMIT = 500
54#: kubectl / client-go failures that say the request did not get through the
55#: tunnel. A match alone never repeats a call: the keeper must also find the
56#: tunnel broken (and reopen it) or already replaced since the call started.
57_TUNNEL_TRANSPORT_MARKERS = (
58 "tls handshake timeout",
59 "unable to connect to the server",
60 "was refused",
61 "connection refused",
62 "connection reset by peer",
63 "i/o timeout",
64)
66_CLUSTER_API_READY_TIMEOUT_SECONDS = 45.0
67_CLUSTER_API_PROBE_TIMEOUT_SECONDS = 8
68_CLUSTER_API_RETRY_SECONDS = 1.0
69_PERMANENT_API_STARTUP_MARKERS = (
70 "certificate signed by unknown authority",
71 "error loading config file",
72 "exec plugin: invalid apiversion",
73 "forbidden",
74 "invalid configuration",
75 "no configuration has been provided",
76 "the server has asked for the client to provide credentials",
77 "tls: failed to verify certificate",
78 "unauthorized",
79 "x509:",
80)
83def _is_permanent_api_startup_error(detail: str) -> bool:
84 normalized = detail.casefold()
85 return any(marker in normalized for marker in _PERMANENT_API_STARTUP_MARKERS)
88def _wait_for_cluster_api(
89 kubectl: KubectlRunner,
90 *,
91 tunnel_process: subprocess.Popen[bytes] | None,
92 timeout_seconds: float = _CLUSTER_API_READY_TIMEOUT_SECONDS,
93 poll_interval_seconds: float = _CLUSTER_API_RETRY_SECONDS,
94) -> None:
95 """Wait until the tunnel can carry an authenticated Kubernetes API request."""
96 if timeout_seconds <= 0:
97 raise ValueError("timeout_seconds must be positive")
98 if poll_interval_seconds <= 0:
99 raise ValueError("poll_interval_seconds must be positive")
101 from cli import ssm_tunnel
103 deadline = time.monotonic() + timeout_seconds
104 attempts = 0
105 last_error = "probe was not attempted"
106 while True:
107 if tunnel_process is not None and (
108 detail := ssm_tunnel.exited_api_tunnel_detail(tunnel_process)
109 ):
110 raise RuntimeError(
111 "SSM tunnel exited before the Kubernetes API became ready: " + detail
112 )
114 remaining = deadline - time.monotonic()
115 if attempts and remaining <= 0:
116 raise RuntimeError(
117 "Kubernetes API did not become ready through the SSM tunnel within "
118 f"{timeout_seconds:.1f}s after {attempts} attempt(s). "
119 f"Last transient error: {last_error[:1000]}"
120 )
121 command_timeout = max(
122 1,
123 min(_CLUSTER_API_PROBE_TIMEOUT_SECONDS, int(max(remaining, 0.0)) + 1),
124 )
125 request_timeout = min(command_timeout, 5)
126 attempts += 1
127 try:
128 returncode, stdout, stderr = kubectl(
129 f"--request-timeout={request_timeout}s",
130 "get",
131 "--raw=/readyz",
132 timeout=command_timeout,
133 )
134 except subprocess.TimeoutExpired as exc:
135 detail = f"kubectl readiness probe timed out after {exc.timeout}s"
136 else:
137 if returncode == 0:
138 return
139 detail = (stderr or stdout).strip() or f"kubectl exited with status {returncode}"
140 if _is_permanent_api_startup_error(detail):
141 raise RuntimeError(
142 "Kubernetes API readiness probe failed with a permanent error: " + detail[:1000]
143 )
144 last_error = detail
146 remaining = deadline - time.monotonic()
147 if remaining <= 0:
148 raise RuntimeError(
149 "Kubernetes API did not become ready through the SSM tunnel within "
150 f"{timeout_seconds:.1f}s after {attempts} attempt(s). "
151 f"Last transient error: {last_error[:1000]}"
152 )
153 time.sleep(min(poll_interval_seconds, remaining))
156class _QuietFormatter:
157 """Adapter for cli formatter callbacks used by the tunnel helpers."""
159 @staticmethod
160 def print_info(message: str) -> None:
161 print(f"[tunnel] {message}")
163 print_success = print_info
164 print_warning = print_info
165 print_error = print_info
168def _kubeconfig_path(kubeconfig_path: Path | None = None) -> Path:
169 return kubeconfig_path if kubeconfig_path is not None else Path.home() / ".kube" / "config"
172def _environment_with_kubeconfig(
173 kubeconfig_path: Path | None,
174 base: Mapping[str, str] | None = None,
175) -> dict[str, str] | None:
176 if kubeconfig_path is None:
177 return dict(base) if base is not None else None
178 environment = dict(base) if base is not None else dict(os.environ)
179 environment["KUBECONFIG"] = str(kubeconfig_path)
180 return environment
183def _update_kubeconfig_command(
184 cluster_name: str,
185 region: str,
186 kubeconfig_path: Path | None,
187) -> list[str]:
188 command = ["aws", "eks", "update-kubeconfig", "--name", cluster_name, "--region", region]
189 if kubeconfig_path is not None:
190 command.extend(("--kubeconfig", str(kubeconfig_path)))
191 return command
194def _validate_and_secure_isolated_kubeconfig(path: Path) -> None:
195 """Require the AWS-written isolated kubeconfig to be a current-user regular file."""
196 metadata = path.lstat()
197 if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
198 raise ValueError(f"Isolated kubeconfig must be a regular file: {path}")
199 if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid():
200 raise PermissionError(f"Isolated kubeconfig is not owned by this user: {path}")
201 if os.name != "nt":
202 path.chmod(0o600)
205def refresh_kubeconfig(
206 cluster_name: str,
207 region: str,
208 *,
209 kubeconfig_path: Path | None = None,
210) -> Path:
211 """Run AWS CLI update-kubeconfig against the selected config file."""
212 subprocess.run(
213 _update_kubeconfig_command(cluster_name, region, kubeconfig_path),
214 check=True,
215 capture_output=True,
216 text=True,
217 env=_environment_with_kubeconfig(kubeconfig_path),
218 shell=False,
219 )
220 path = _kubeconfig_path(kubeconfig_path)
221 if kubeconfig_path is not None:
222 _validate_and_secure_isolated_kubeconfig(path)
223 return path
226def update_and_point_kubeconfig_at_tunnel(
227 cluster_name: str,
228 region: str,
229 server: str,
230 tls_server_name: str,
231 *,
232 kubeconfig_path: Path | None = None,
233) -> None:
234 """Refresh kubeconfig, point it at the tunnel, and preserve real TLS SNI."""
235 path = refresh_kubeconfig(
236 cluster_name,
237 region,
238 kubeconfig_path=kubeconfig_path,
239 )
240 config = yaml.safe_load(path.read_text(encoding="utf-8"))
241 if not isinstance(config, dict) or not isinstance(config.get("clusters"), list):
242 raise ValueError(f"Kubeconfig has no cluster list: {path}")
243 expected_suffix = f"cluster/{cluster_name}"
244 matched = False
245 for entry in config["clusters"]:
246 if not isinstance(entry, dict) or not str(entry.get("name", "")).endswith(expected_suffix):
247 continue
248 cluster = entry.get("cluster")
249 if not isinstance(cluster, dict):
250 raise ValueError(f"Kubeconfig cluster entry is malformed: {path}")
251 cluster["server"] = server
252 cluster["tls-server-name"] = tls_server_name
253 matched = True
254 if not matched:
255 raise ValueError(f"Kubeconfig did not contain the requested cluster: {cluster_name}")
256 path.write_text(yaml.safe_dump(config), encoding="utf-8")
257 if kubeconfig_path is not None:
258 _validate_and_secure_isolated_kubeconfig(path)
261def ensure_cluster_access_entry(
262 repo_root: Path,
263 region: str,
264 *,
265 kubeconfig_path: Path | None = None,
266 gco_command: tuple[str, ...] = ("gco",),
267) -> None:
268 """Grant cluster-admin through an explicitly selected GCO checkout."""
269 if not gco_command or any(not isinstance(part, str) or not part for part in gco_command):
270 raise ValueError("gco_command must be a non-empty argv prefix")
271 result = subprocess.run(
272 [*gco_command, "stacks", "access", "--region", region],
273 cwd=repo_root,
274 capture_output=True,
275 text=True,
276 env=_environment_with_kubeconfig(kubeconfig_path),
277 shell=False,
278 check=False,
279 )
280 if result.returncode != 0:
281 raise RuntimeError(f"gco stacks access failed for {region}: {result.stderr.strip()[:500]}")
284class TunnelUnavailableError(RuntimeError):
285 """The session's SSM tunnel is broken and could not be reopened."""
288def is_tunnel_transport_error(detail: str) -> bool:
289 """True when kubectl's error says the request did not get through the tunnel."""
290 normalized = detail.casefold()
291 return any(marker in normalized for marker in _TUNNEL_TRANSPORT_MARKERS)
294def _tunnel_handshake_error(port: int, server_name: str, timeout: float) -> str | None:
295 """``None`` when a TLS handshake completes through the local tunnel port, else why not.
297 A stalled Session Manager session still accepts the local TCP connection
298 and never relays the server's answer, so the handshake times out: the
299 state this probe exists to catch. The probe sends no request. It does not
300 need to trust the API server's certificate either — the cluster CA is not
301 in the system store — because a certificate the client could not verify
302 has, by then, come back through the tunnel.
303 """
304 context = ssl.create_default_context()
305 context.minimum_version = ssl.TLSVersion.TLSv1_2
306 try:
307 with (
308 socket.create_connection(("127.0.0.1", port), timeout=timeout) as connection,
309 context.wrap_socket(connection, server_hostname=server_name),
310 ):
311 return None
312 except ssl.SSLCertVerificationError:
313 return None
314 except OSError as exc:
315 return f"{type(exc).__name__}: {exc}"
318def _utc_timestamp() -> str:
319 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
322class _TunnelKeeper:
323 """Keep one session's SSM tunnel carrying the Kubernetes API.
325 "The session process runs" and "the port accepts" prove nothing about a
326 Session Manager port-forward; only a TLS handshake answered through it
327 does. The keeper reopens the session on the same local port through the
328 same instance and counts every successful reopen in ``generation``, so a
329 call can tell whether the tunnel it failed on has been replaced since.
330 Every attempt is appended to ``events`` for the run's evidence.
331 """
333 def __init__(
334 self,
335 *,
336 instance_id: str,
337 endpoint: str,
338 local_port: int,
339 region: str,
340 server_name: str,
341 process: subprocess.Popen[bytes] | None,
342 wait_ready: Callable[[subprocess.Popen[bytes]], None],
343 formatter: Any,
344 events: list[dict[str, Any]],
345 ) -> None:
346 self._instance_id = instance_id
347 self._endpoint = endpoint
348 self._local_port = local_port
349 self._region = region
350 self._server_name = server_name
351 self._original = process
352 self.process = process
353 self._wait_ready = wait_ready
354 self._formatter = formatter
355 self.events = events
356 self.generation = 0
357 self._lock = threading.Lock()
358 self._closed = threading.Event()
359 self._thread: threading.Thread | None = None
360 self._failures = 0
361 self._retry_after = 0.0
362 self._last_error = ""
364 # -- health -------------------------------------------------------------
366 def _exit_detail(self) -> str | None:
367 process = self.process
368 if process is None:
369 return None
370 code = process.poll()
371 return None if code is None else f"the SSM session exited with code {code}"
373 def _broken_reason(self) -> str | None:
374 if (exited := self._exit_detail()) is not None:
375 return exited
376 error = _tunnel_handshake_error(
377 self._local_port, self._server_name, _TUNNEL_HANDSHAKE_TIMEOUT_SECONDS
378 )
379 if error is None:
380 return None
381 return (
382 f"no TLS handshake through 127.0.0.1:{self._local_port} within "
383 f"{_TUNNEL_HANDSHAKE_TIMEOUT_SECONDS:g}s ({error})"
384 )
386 # -- reopening (callers hold the lock) ----------------------------------
388 def _reopen(self, reason: str) -> bool:
389 """Replace the SSM session; True once the Kubernetes API answers through it."""
390 if self._closed.is_set():
391 # close() stops only the session it can see; never open one after it.
392 self._last_error = "the cluster session is closing"
393 return False
394 if time.monotonic() < self._retry_after:
395 # _last_error still describes the failure that started the backoff.
396 return False
397 from cli import ssm_tunnel
399 event: dict[str, Any] = {
400 "at": _utc_timestamp(),
401 "reason": reason[:_TUNNEL_EVENT_TEXT_LIMIT],
402 }
403 self.events.append(event)
404 self._formatter.print_warning(f"Reopening the SSM tunnel: {reason}")
405 stale, self.process = self.process, None
406 if stale is not None:
407 try:
408 ssm_tunnel.stop_api_tunnel(stale)
409 except RuntimeError as exc:
410 event["stop_error"] = str(exc)[:_TUNNEL_EVENT_TEXT_LIMIT]
411 try:
412 self.process = ssm_tunnel.start_api_tunnel(
413 self._instance_id, self._endpoint, self._local_port, self._region
414 )
415 self._wait_ready(self.process)
416 except Exception as exc: # any failure leaves the tunnel down; say so and back off
417 self._last_error = f"{type(exc).__name__}: {exc}"[:_TUNNEL_EVENT_TEXT_LIMIT]
418 event["result"] = "failed"
419 event["error"] = self._last_error
420 self._retry_after = time.monotonic() + _TUNNEL_REOPEN_BACKOFF_SECONDS
421 self._formatter.print_error(f"Could not reopen the SSM tunnel: {self._last_error}")
422 return False
423 self.generation += 1
424 event["result"] = "reopened"
425 self._formatter.print_success("SSM tunnel reopened; the Kubernetes API answers again.")
426 return True
428 # -- entry points -------------------------------------------------------
430 def recover(self, since: int) -> bool:
431 """After a call made at generation ``since`` failed on the transport.
433 True when repeating the call can help: the tunnel was replaced after
434 the call started, or it is broken now and has just been reopened. A
435 healthy tunnel returns False, so a failure that was not the tunnel's
436 is reported as it happened rather than repeated.
437 """
438 with self._lock:
439 if self.generation != since:
440 return True
441 reason = self._broken_reason()
442 if reason is None:
443 return False
444 return self._reopen(f"a call through the tunnel failed and {reason}")
446 def ensure(self) -> None:
447 """Reopen a broken tunnel now; raise :class:`TunnelUnavailableError` if that fails."""
448 with self._lock:
449 reason = self._broken_reason()
450 if reason is None or self._reopen(reason):
451 return
452 raise TunnelUnavailableError(
453 f"the SSM tunnel to the Kubernetes API is broken ({reason}) and could not "
454 f"be reopened: {self._last_error}"
455 )
457 def check(self) -> None:
458 """One watchdog pass: reopen an exited session, or after consecutive failed handshakes."""
459 with self._lock:
460 reason = self._broken_reason()
461 if reason is None:
462 self._failures = 0
463 return
464 self._failures += 1
465 if self._failures < _TUNNEL_FAILURES_BEFORE_REOPEN and self._exit_detail() is None:
466 return
467 self._failures = 0
468 self._reopen(f"watchdog: {reason}")
470 # -- watchdog lifecycle -------------------------------------------------
472 def _watch(self) -> None:
473 while not self._closed.wait(_TUNNEL_CHECK_INTERVAL_SECONDS):
474 try:
475 self.check()
476 except Exception as exc: # the watchdog must outlive any one bad probe
477 self._formatter.print_warning(
478 f"SSM tunnel watchdog check failed: {type(exc).__name__}: {exc}"
479 )
481 def start(self) -> None:
482 self._thread = threading.Thread(target=self._watch, name="gco-tunnel-watchdog", daemon=True)
483 self._thread.start()
485 def close(self) -> None:
486 """Stop the watchdog, then any session it opened (the original is its opener's)."""
487 self._closed.set()
488 if self._thread is not None:
489 self._thread.join()
490 from cli import ssm_tunnel
492 with self._lock:
493 process = self.process
494 if process is None or process is self._original:
495 return
496 try:
497 ssm_tunnel.stop_api_tunnel(process)
498 except RuntimeError as exc:
499 self._formatter.print_error(f"Could not stop the reopened SSM tunnel: {exc}")
502class SessionKubectl:
503 """The kubectl runner :func:`cluster_session` yields.
505 Calling it runs kubectl against the session's kubeconfig and returns
506 ``(code, stdout, stderr)``; a non-zero exit is data, never an exception.
507 With a tunnel keeper, a call that fails on the tunnel's transport is
508 repeated once after the keeper has reopened a broken tunnel.
509 """
511 def __init__(self, kubeconfig_path: Path | None, keeper: _TunnelKeeper | None = None) -> None:
512 self._kubeconfig_path = kubeconfig_path
513 self.keeper = keeper
515 def run_once(self, *args: str, timeout: float = 120, **kwargs: Any) -> tuple[int, str, str]:
516 """One kubectl invocation, never repeated."""
517 if kwargs.pop("shell", False):
518 raise ValueError("cluster_session kubectl does not allow shell execution")
519 command = ["kubectl"]
520 if self._kubeconfig_path is not None:
521 command.extend(("--kubeconfig", str(self._kubeconfig_path)))
522 command.extend(args)
523 caller_environment = kwargs.pop("env", None)
524 environment = _environment_with_kubeconfig(self._kubeconfig_path, caller_environment)
525 # Callers branch on the code, so check=True would only turn data into
526 # an exception here.
527 kwargs.pop("check", None)
528 result = subprocess.run(
529 command,
530 capture_output=True,
531 text=True,
532 timeout=timeout,
533 env=environment,
534 shell=False,
535 check=False,
536 **kwargs,
537 )
538 return result.returncode, result.stdout, result.stderr
540 def __call__(self, *args: str, timeout: float = 120, **kwargs: Any) -> tuple[int, str, str]:
541 return self.through_tunnel(lambda: self.run_once(*args, timeout=timeout, **kwargs))
543 def environment(self) -> dict[str, str] | None:
544 """``os.environ`` with ``KUBECONFIG`` set to this session's kubeconfig, or ``None``."""
545 return _environment_with_kubeconfig(self._kubeconfig_path)
547 def through_tunnel(
548 self,
549 call: Callable[[], tuple[int, str, str]],
550 *,
551 safe_to_repeat: Callable[[], bool] | None = None,
552 ) -> tuple[int, str, str]:
553 """Run ``call``; repeat it once when it failed on a tunnel the keeper then reopened.
555 ``call`` is anything that crosses the tunnel and returns ``(code,
556 stdout, stderr)`` — kubectl itself, or a ``gco`` command that shells
557 out to it. ``safe_to_repeat`` is asked after the tunnel is back, for a
558 call whose repetition could duplicate work the first attempt did.
559 """
560 keeper = self.keeper
561 if keeper is None:
562 return call()
563 generation = keeper.generation
564 code, stdout, stderr = call()
565 if (
566 code != 0
567 and is_tunnel_transport_error(f"{stderr}\n{stdout}")
568 and keeper.recover(generation)
569 and (safe_to_repeat is None or safe_to_repeat())
570 ):
571 return call()
572 return code, stdout, stderr
574 def ensure_tunnel(self) -> None:
575 """Reopen a broken tunnel now (a no-op without one); raise when it cannot be."""
576 if self.keeper is not None:
577 self.keeper.ensure()
580def through_tunnel(
581 kubectl: KubectlRunner,
582 call: Callable[[], tuple[int, str, str]],
583 *,
584 safe_to_repeat: Callable[[], bool] | None = None,
585) -> tuple[int, str, str]:
586 """:meth:`SessionKubectl.through_tunnel` for a session runner; ``call()`` for any other."""
587 if isinstance(kubectl, SessionKubectl):
588 return kubectl.through_tunnel(call, safe_to_repeat=safe_to_repeat)
589 return call()
592def ensure_tunnel(kubectl: KubectlRunner) -> None:
593 """:meth:`SessionKubectl.ensure_tunnel` for a session runner; a no-op for any other."""
594 if isinstance(kubectl, SessionKubectl):
595 kubectl.ensure_tunnel()
598def session_environment(kubectl: KubectlRunner) -> dict[str, str] | None:
599 """The environment for a ``gco`` process that must reach the cluster the way kubectl does.
601 A session runner with its own kubeconfig hands it on through
602 ``KUBECONFIG``, over whatever the caller's environment exports, so a
603 ``gco`` command that shells out to kubectl (``gco jobs submit-direct``)
604 uses the file the tunnel is pinned in. ``None`` (inherit) for a session on
605 the default kubeconfig and for any other runner.
606 """
607 if isinstance(kubectl, SessionKubectl):
608 return kubectl.environment()
609 return None
612@contextmanager
613def cluster_session(
614 repo_root: Path,
615 cluster_name: str,
616 region: str,
617 *,
618 kubeconfig_path: Path | None = None,
619 gco_command: tuple[str, ...] = ("gco",),
620 bastion_ttl_minutes: int | None = None,
621 tunnel_events: list[dict[str, Any]] | None = None,
622) -> Iterator[KubectlRunner]:
623 """Access entry plus tunnel for one region; optionally isolate kubeconfig.
625 The session tunnels through an ephemeral bastion whenever the cluster has
626 private endpoint access (``PUBLIC_AND_PRIVATE`` included, so a public
627 endpoint behind a CIDR allowlist never decides the run), and uses a
628 public-only endpoint directly. ``bastion_ttl_minutes`` sizes the ephemeral
629 bastion's self-termination backstop (the bastion's default otherwise);
630 ``tunnel_events`` collects every tunnel reopen the session's keeper
631 attempted.
632 """
633 from cli import cluster_tunnel
635 formatter = _QuietFormatter()
636 ensure_cluster_access_entry(
637 repo_root,
638 region,
639 kubeconfig_path=kubeconfig_path,
640 gco_command=gco_command,
641 )
642 with cluster_tunnel.open_api_server_tunnel(
643 formatter,
644 cluster=cluster_name,
645 region=region,
646 via_ssm=cluster_tunnel.AUTO_BASTION,
647 assume_yes=True,
648 bastion_ttl_minutes=bastion_ttl_minutes,
649 allow_public_fallback=True,
650 ) as session:
651 if session.active and session.server and session.tls_server_name:
652 update_and_point_kubeconfig_at_tunnel(
653 cluster_name,
654 region,
655 session.server,
656 session.tls_server_name,
657 kubeconfig_path=kubeconfig_path,
658 )
659 else:
660 refresh_kubeconfig(
661 cluster_name,
662 region,
663 kubeconfig_path=kubeconfig_path,
664 )
666 kubectl = SessionKubectl(kubeconfig_path)
667 process = getattr(session, "process", None)
668 if session.active:
669 _wait_for_cluster_api(kubectl.run_once, tunnel_process=process)
670 instance_id = getattr(session, "instance_id", None)
671 plan = getattr(session, "plan", None)
672 if not (session.active and instance_id and plan is not None and session.tls_server_name):
673 yield kubectl
674 return
675 keeper = _TunnelKeeper(
676 instance_id=instance_id,
677 endpoint=plan.endpoint,
678 local_port=plan.local_port,
679 region=region,
680 server_name=session.tls_server_name,
681 process=process,
682 wait_ready=lambda reopened: _wait_for_cluster_api(
683 kubectl.run_once, tunnel_process=reopened
684 ),
685 formatter=formatter,
686 events=[] if tunnel_events is None else tunnel_events,
687 )
688 kubectl.keeper = keeper
689 keeper.start()
690 try:
691 yield kubectl
692 finally:
693 keeper.close()