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:
@@ -64,6 +64,11 @@ def main() -> None:
|
||||
# sunbeam bootstrap
|
||||
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()
|
||||
|
||||
if args.verb is None:
|
||||
@@ -123,6 +128,10 @@ def main() -> None:
|
||||
from sunbeam.gitea import cmd_bootstrap
|
||||
cmd_bootstrap()
|
||||
|
||||
elif args.verb == "k8s":
|
||||
from sunbeam.kube import cmd_k8s
|
||||
sys.exit(cmd_k8s(args.kubectl_args))
|
||||
|
||||
else:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
@@ -93,11 +93,13 @@ def create_secret(ns: str, name: str, **literals) -> None:
|
||||
"--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)."""
|
||||
r = run_tool("kubectl", "--context=sunbeam", "exec", "-n", ns, pod,
|
||||
"--", *cmd,
|
||||
capture_output=True, text=True, check=False)
|
||||
args = ["kubectl", "--context=sunbeam", "exec", "-n", ns, pod]
|
||||
if container:
|
||||
args += ["-c", container]
|
||||
args += ["--", *cmd]
|
||||
r = run_tool(*args, capture_output=True, text=True, check=False)
|
||||
return r.returncode, r.stdout.strip()
|
||||
|
||||
|
||||
@@ -115,6 +117,15 @@ def get_domain() -> str:
|
||||
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:
|
||||
"""Run kustomize build --enable-helm and apply domain substitution."""
|
||||
r = run_tool(
|
||||
|
||||
Reference in New Issue
Block a user