feat: add sunbeam k8s kubectl passthrough; fix kube_exec container arg

kube.py: kube_exec now accepts an optional container= kwarg so callers
can target a specific container in Linkerd-injected pods (where exec
would otherwise land in the linkerd-proxy sidecar instead of the app).
Used by check_valkey (container="valkey") and check_openbao
(container="openbao").

kube.py + cli.py: new cmd_k8s / sunbeam k8s verb — transparent
kubectl --context=sunbeam passthrough for one-off cluster operations.
Returns kubectl's exit code directly.
This commit is contained in:
2026-03-03 00:57:48 +00:00
parent fb3fd93f0f
commit 352f0b6869
2 changed files with 24 additions and 4 deletions

View File

@@ -64,6 +64,11 @@ def main() -> None:
# sunbeam bootstrap # sunbeam bootstrap
sub.add_parser("bootstrap", help="Create Gitea orgs/repos; set up Lima registry") sub.add_parser("bootstrap", help="Create Gitea orgs/repos; set up Lima registry")
# sunbeam k8s [kubectl args...] — transparent kubectl --context=sunbeam wrapper
p_k8s = sub.add_parser("k8s", help="kubectl --context=sunbeam passthrough")
p_k8s.add_argument("kubectl_args", nargs=argparse.REMAINDER,
help="arguments forwarded verbatim to kubectl")
args = parser.parse_args() args = parser.parse_args()
if args.verb is None: if args.verb is None:
@@ -123,6 +128,10 @@ def main() -> None:
from sunbeam.gitea import cmd_bootstrap from sunbeam.gitea import cmd_bootstrap
cmd_bootstrap() cmd_bootstrap()
elif args.verb == "k8s":
from sunbeam.kube import cmd_k8s
sys.exit(cmd_k8s(args.kubectl_args))
else: else:
parser.print_help() parser.print_help()
sys.exit(1) sys.exit(1)

View File

@@ -93,11 +93,13 @@ def create_secret(ns: str, name: str, **literals) -> None:
"--field-manager=sunbeam", "-f", "-", input=manifest) "--field-manager=sunbeam", "-f", "-", input=manifest)
def kube_exec(ns: str, pod: str, *cmd: str) -> tuple[int, str]: def kube_exec(ns: str, pod: str, *cmd: str, container: str | None = None) -> tuple[int, str]:
"""Run a command inside a pod. Returns (returncode, stdout).""" """Run a command inside a pod. Returns (returncode, stdout)."""
r = run_tool("kubectl", "--context=sunbeam", "exec", "-n", ns, pod, args = ["kubectl", "--context=sunbeam", "exec", "-n", ns, pod]
"--", *cmd, if container:
capture_output=True, text=True, check=False) args += ["-c", container]
args += ["--", *cmd]
r = run_tool(*args, capture_output=True, text=True, check=False)
return r.returncode, r.stdout.strip() return r.returncode, r.stdout.strip()
@@ -115,6 +117,15 @@ def get_domain() -> str:
return f"{ip}.sslip.io" return f"{ip}.sslip.io"
def cmd_k8s(kubectl_args: list[str]) -> int:
"""Transparent kubectl --context=sunbeam passthrough. Returns kubectl's exit code."""
from sunbeam.tools import ensure_tool
import os
bin_path = ensure_tool("kubectl")
r = subprocess.run([str(bin_path), "--context=sunbeam", *kubectl_args])
return r.returncode
def kustomize_build(overlay: Path, domain: str) -> str: def kustomize_build(overlay: Path, domain: str) -> str:
"""Run kustomize build --enable-helm and apply domain substitution.""" """Run kustomize build --enable-helm and apply domain substitution."""
r = run_tool( r = run_tool(