Coverage for scripts / live_release_validation / checks / cluster.py: 100.00%

1"""kubectl plumbing shared by the cluster-facing checks. 

2 

3The ``platform-workloads`` and ``network-posture`` actions read and create 

4Kubernetes objects on every deployed Region's cluster. Both reach the private 

5API endpoint the way the ``inference`` action does — access entry, SSM tunnel, 

6and the isolated ``kubeconfig`` inside the private report directory 

7(``scripts.example_job_validation.kube.cluster_session``) — and both need 

8fail-closed JSON reads that distinguish "the object is absent" from "the read 

9broke". This module owns that plumbing so neither check re-implements it. 

10""" 

11 

12from __future__ import annotations 

13 

14import json 

15import sys 

16from collections.abc import Callable, Iterator 

17from contextlib import contextmanager 

18from typing import Any 

19 

20from scripts.example_job_validation import kube 

21 

22from ..models import RunContext 

23 

24#: ``kubectl(*argv, timeout=..., **subprocess_kwargs) -> (returncode, stdout, stderr)`` 

25KubectlRunner = Callable[..., tuple[int, str, str]] 

26 

27_OUTPUT_LIMIT = 1_000 

28 

29 

30class KubectlError(RuntimeError): 

31 """A kubectl invocation the checks could not interpret.""" 

32 

33 

34def _truncated(value: str) -> str: 

35 return value if len(value) <= _OUTPUT_LIMIT else value[-_OUTPUT_LIMIT:] 

36 

37 

38def _not_found(stderr: str) -> bool: 

39 lowered = stderr.casefold() 

40 return "notfound" in lowered or "not found" in lowered 

41 

42 

43def kubectl_json( 

44 kubectl: KubectlRunner, 

45 record: dict[str, Any], 

46 *arguments: str, 

47 timeout: float, 

48) -> Any | None: 

49 """Run ``kubectl <arguments> --output json``; ``None`` when the object is absent. 

50 

51 Any other non-zero exit, and any unparsable output, is recorded in 

52 ``record["last_kubectl_error"]`` (bounded) and raised, so a broken tunnel 

53 or a revoked permission can never read as "the object was not there". 

54 """ 

55 code, stdout, stderr = kubectl(*arguments, "--output", "json", timeout=timeout) 

56 if code != 0: 

57 if _not_found(stderr): 

58 return None 

59 record["last_kubectl_error"] = { 

60 "argv": list(arguments), 

61 "returncode": code, 

62 "stdout": _truncated(stdout), 

63 "stderr": _truncated(stderr), 

64 } 

65 raise KubectlError( 

66 f"kubectl {' '.join(arguments[:2])} failed with exit {code}; " 

67 "the checkpoint holds the output" 

68 ) 

69 try: 

70 return json.loads(stdout) 

71 except json.JSONDecodeError as exc: 

72 record["last_kubectl_error"] = { 

73 "argv": list(arguments), 

74 "returncode": code, 

75 "stdout": _truncated(stdout), 

76 "error": f"{type(exc).__name__}: {exc}", 

77 } 

78 raise KubectlError( 

79 f"kubectl {' '.join(arguments[:2])} returned invalid JSON; " 

80 "the checkpoint holds the output" 

81 ) from None 

82 

83 

84@contextmanager 

85def cluster_kubectl(ctx: RunContext, region: str) -> Iterator[KubectlRunner]: 

86 """Yield a tunnelled kubectl for one Region's cluster via the isolated kubeconfig. 

87 

88 The kubeconfig is the single private-directory file the runner already 

89 accounts for (``RunSettings.kubeconfig_path``); Regions are visited one at 

90 a time, and each session re-points it at its own cluster. 

91 """ 

92 settings = ctx.settings 

93 kubeconfig_path = settings.kubeconfig_path 

94 if kubeconfig_path.parent != settings.report_dir: 

95 raise KubectlError("isolated kubeconfig escaped the private report dir") 

96 cluster_name = f"{ctx.config.project_name}-{region}" 

97 with kube.cluster_session( 

98 settings.repo_root, 

99 cluster_name, 

100 region, 

101 kubeconfig_path=kubeconfig_path, 

102 gco_command=(sys.executable, "-m", "cli.main"), 

103 ) as kubectl: 

104 yield kubectl