158 lines
5.0 KiB
Python
158 lines
5.0 KiB
Python
"""Правка kubeconfig kind для доступа к API с хоста (после create из контейнера).
|
||
|
||
Kind внутри Docker видит другой адрес apiserver; на хосте нужен 127.0.0.1:<порт>
|
||
из проброса `docker port <cluster>-control-plane 6443/tcp` (Podman — тот же клиент
|
||
`docker` к docker-совместимому сокету).
|
||
|
||
Автор: Сергей Антропов
|
||
Сайт: https://devops.org.ru
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
logger = logging.getLogger("kind_k8s.kubeconfig_patch")
|
||
|
||
|
||
def _container_cli() -> str:
|
||
"""CLI для `port` (обычно docker к сокету Docker или Podman)."""
|
||
return (os.environ.get("CONTAINER_CLI") or "docker").strip() or "docker"
|
||
|
||
|
||
def _control_plane_container_name(cluster_name: str) -> str:
|
||
return f"{cluster_name}-control-plane"
|
||
|
||
|
||
def _parse_docker_port_line(line: str) -> tuple[str, str] | None:
|
||
"""Строка вида '0.0.0.0:32768' или '127.0.0.1:32768' -> (host, port)."""
|
||
line = line.strip()
|
||
if ":" not in line:
|
||
return None
|
||
# IPv6 [::]:port
|
||
if line.startswith("["):
|
||
rb = line.rfind("]")
|
||
if rb == -1:
|
||
return None
|
||
host = line[1:rb]
|
||
rest = line[rb + 1 :].lstrip(":")
|
||
if not rest.isdigit():
|
||
return None
|
||
return host, rest
|
||
host, _, port = line.rpartition(":")
|
||
if not port.isdigit():
|
||
return None
|
||
host = host.strip()
|
||
return host, port
|
||
|
||
|
||
def _host_bind_for_kubeconfig(host: str) -> str:
|
||
if host in ("0.0.0.0", "::", ""):
|
||
return "127.0.0.1"
|
||
if host == "[::]":
|
||
return "127.0.0.1"
|
||
return host
|
||
|
||
|
||
def get_apiserver_host_port(cluster_name: str) -> tuple[str, str] | None:
|
||
"""Узнать (host, port) с хоста для доступа к apiserver."""
|
||
cli = _container_cli()
|
||
ctr = _control_plane_container_name(cluster_name)
|
||
p = subprocess.run(
|
||
[cli, "port", ctr, "6443/tcp"],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if p.returncode != 0:
|
||
logger.info(
|
||
"Не удалось %s port %s 6443/tcp: %s",
|
||
cli,
|
||
ctr,
|
||
(p.stderr or p.stdout or "").strip(),
|
||
)
|
||
return None
|
||
for raw in (p.stdout or "").splitlines():
|
||
parsed = _parse_docker_port_line(raw)
|
||
if not parsed:
|
||
continue
|
||
h, port = parsed
|
||
return _host_bind_for_kubeconfig(h), port
|
||
return None
|
||
|
||
|
||
def _kubeconfig_cluster_name(kube_path: Path, logical_name: str) -> str:
|
||
"""Имя блока cluster в kubeconfig (обычно kind-<имя>)."""
|
||
p = subprocess.run(
|
||
[
|
||
"kubectl",
|
||
"--kubeconfig",
|
||
str(kube_path),
|
||
"config",
|
||
"view",
|
||
"-o",
|
||
'jsonpath={range .clusters[*]}{.name}{"\n"}{end}',
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
names = [x.strip() for x in (p.stdout or "").splitlines() if x.strip()]
|
||
want = f"kind-{logical_name}"
|
||
if want in names:
|
||
return want
|
||
for n in names:
|
||
if n.startswith("kind-"):
|
||
return n
|
||
return want
|
||
|
||
|
||
def patch_kubeconfig_server_for_host(
|
||
*,
|
||
cluster_name: str,
|
||
kube_path: Path,
|
||
) -> bool:
|
||
"""
|
||
Подставить в kubeconfig server=https://<хост>:<порт> для доступа с хоста.
|
||
|
||
Порт берётся из ``docker port`` / аналога к сокету; для Podman — тот же клиент
|
||
при ``DOCKER_HOST`` на podman.sock.
|
||
"""
|
||
hp = get_apiserver_host_port(cluster_name)
|
||
if not hp:
|
||
print(
|
||
"Предупреждение: не удалось получить порт apiserver с хоста; "
|
||
"kubeconfig оставлен как выдал kind (с хоста может не открываться).",
|
||
)
|
||
return False
|
||
host, port = hp
|
||
server = f"https://{host}:{port}"
|
||
cluster_id = _kubeconfig_cluster_name(kube_path, cluster_name)
|
||
p = subprocess.run(
|
||
[
|
||
"kubectl",
|
||
"--kubeconfig",
|
||
str(kube_path),
|
||
"config",
|
||
"set-cluster",
|
||
cluster_id,
|
||
f"--server={server}",
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if p.returncode != 0:
|
||
err = (p.stderr or p.stdout or "").strip()
|
||
print(f"Предупреждение: kubectl config set-cluster не удался: {err}")
|
||
return False
|
||
print(f"Kubeconfig для хоста: apiserver → {server}")
|
||
return True
|
||
|
||
|
||
def should_patch_after_create() -> bool:
|
||
"""Патчить после create, если задано явно или kind шёл из контейнера."""
|
||
if os.environ.get("KIND_K8S_PATCH_KUBECONFIG", "").strip().lower() in ("1", "true", "yes", "да"):
|
||
return True
|
||
return os.environ.get("KIND_K8S_IN_CONTAINER", "").strip() == "1"
|