mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 12:03:59 +00:00
317 lines
8.6 KiB
Python
317 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
_LogFn = Callable[[str], None]
|
|
|
|
|
|
def _log(log: _LogFn | None, msg: str) -> None:
|
|
if log:
|
|
log(msg)
|
|
|
|
|
|
def _merge_env(env: dict | None) -> dict:
|
|
merged = dict(os.environ)
|
|
if env:
|
|
merged.update(env)
|
|
return merged
|
|
|
|
|
|
def _run(
|
|
args: list[str],
|
|
*,
|
|
env: dict | None = None,
|
|
input_text: str | None = None,
|
|
timeout: int = 300,
|
|
check: bool = False,
|
|
) -> subprocess.CompletedProcess:
|
|
res = subprocess.run(
|
|
args,
|
|
input=input_text,
|
|
capture_output=True,
|
|
text=True,
|
|
env=_merge_env(env),
|
|
timeout=timeout,
|
|
)
|
|
if check and res.returncode != 0:
|
|
raise RuntimeError(
|
|
f"Command failed ({res.returncode}): {' '.join(args)}\n"
|
|
f"stdout: {res.stdout}\n"
|
|
f"stderr: {res.stderr}"
|
|
)
|
|
return res
|
|
|
|
|
|
def _kubectl(
|
|
args: list[str],
|
|
*,
|
|
env: dict | None = None,
|
|
input_text: str | None = None,
|
|
timeout: int = 300,
|
|
check: bool = False,
|
|
) -> subprocess.CompletedProcess:
|
|
# KUBECONTEXT is not a real kubectl env var — extract it and pass as --context flag.
|
|
ctx = str((env or {}).get("KUBECONTEXT") or "").strip()
|
|
cmd = ["kubectl"]
|
|
if ctx:
|
|
cmd.extend(["--context", ctx])
|
|
cmd.extend(args)
|
|
return _run(cmd, env=env, input_text=input_text, timeout=timeout, check=check)
|
|
|
|
|
|
def _helm(
|
|
args: list[str],
|
|
*,
|
|
env: dict | None = None,
|
|
timeout: int = 300,
|
|
check: bool = False,
|
|
) -> subprocess.CompletedProcess:
|
|
# KUBECONTEXT is not a real helm env var — extract it and pass as --kube-context flag.
|
|
ctx = str((env or {}).get("KUBECONTEXT") or "").strip()
|
|
cmd = ["helm"]
|
|
if ctx:
|
|
cmd.extend(["--kube-context", ctx])
|
|
cmd.extend(args)
|
|
return _run(cmd, env=env, timeout=timeout, check=check)
|
|
|
|
|
|
def _k3d(
|
|
args: list[str],
|
|
*,
|
|
env: dict | None = None,
|
|
timeout: int = 120,
|
|
check: bool = False,
|
|
) -> subprocess.CompletedProcess:
|
|
return _run(["k3d", *args], env=env, timeout=timeout, check=check)
|
|
|
|
|
|
def _docker(
|
|
args: list[str],
|
|
*,
|
|
env: dict | None = None,
|
|
timeout: int = 120,
|
|
check: bool = False,
|
|
) -> subprocess.CompletedProcess:
|
|
return _run(["docker", *args], env=env, timeout=timeout, check=check)
|
|
|
|
|
|
def _detect_mode(mode: str | None, env: dict | None) -> str:
|
|
if mode:
|
|
return str(mode).strip()
|
|
if env:
|
|
for key in ("KNOE_MODE", "DEPLOYMENT_MODE"):
|
|
value = str(env.get(key) or "").strip()
|
|
if value:
|
|
return value
|
|
return "k3s"
|
|
|
|
|
|
def _namespace(namespace: str | None, env: dict | None, default: str = "default") -> str:
|
|
raw = (
|
|
(namespace or "").strip()
|
|
or str((env or {}).get("SERVICE_NAMESPACE") or "").strip()
|
|
or str((env or {}).get("NAMESPACE") or "").strip()
|
|
or default
|
|
)
|
|
return raw
|
|
|
|
|
|
def _registry_namespace(namespace: str | None, env: dict | None) -> str:
|
|
service_ns = str((env or {}).get("SERVICE_NAMESPACE") or "").strip()
|
|
raw = (
|
|
(namespace or "").strip()
|
|
or str((env or {}).get("REGISTRY_NAMESPACE") or "").strip()
|
|
or service_ns
|
|
or str((env or {}).get("NAMESPACE") or "").strip()
|
|
or "knoe-system"
|
|
)
|
|
if raw == "default" and service_ns:
|
|
return service_ns
|
|
return raw
|
|
|
|
|
|
def _ensure_namespace(namespace: str, env: dict | None = None) -> None:
|
|
res = _kubectl(["get", "namespace", namespace], env=env, timeout=30)
|
|
if res.returncode == 0:
|
|
return
|
|
_kubectl(["create", "namespace", namespace], env=env, timeout=60, check=True)
|
|
|
|
|
|
def _wait_rollout(
|
|
kind: str,
|
|
name: str,
|
|
namespace: str,
|
|
*,
|
|
env: dict | None = None,
|
|
timeout: str = "300s",
|
|
) -> None:
|
|
_kubectl(
|
|
[
|
|
"-n",
|
|
namespace,
|
|
"rollout",
|
|
"status",
|
|
f"{kind}/{name}",
|
|
f"--timeout={timeout}",
|
|
],
|
|
env=env,
|
|
timeout=360,
|
|
check=True,
|
|
)
|
|
|
|
|
|
def _manifest_path(project_root: str | Path, *segments: str) -> Path:
|
|
return Path(project_root).joinpath(*segments)
|
|
|
|
|
|
def _exists(kind: str, name: str, namespace: str, env: dict | None = None) -> bool:
|
|
res = _kubectl(["-n", namespace, "get", kind, name], env=env, timeout=20)
|
|
return res.returncode == 0
|
|
|
|
|
|
def _to_bool(value: str | bool | None, default: bool = False) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
if value is None:
|
|
return default
|
|
return str(value).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _prune_named_workload_other_namespaces(
|
|
*,
|
|
kind: str,
|
|
name: str,
|
|
target_namespace: str,
|
|
label_selector: str,
|
|
env: dict | None = None,
|
|
log: _LogFn | None = None,
|
|
) -> None:
|
|
listed = _kubectl(
|
|
["get", kind, "-A", "-l", label_selector, "-o", "json"],
|
|
env=env,
|
|
timeout=45,
|
|
)
|
|
if listed.returncode != 0 or not listed.stdout.strip():
|
|
return
|
|
try:
|
|
payload = json.loads(listed.stdout)
|
|
except Exception:
|
|
return
|
|
|
|
for item in payload.get("items", []):
|
|
meta = item.get("metadata") or {}
|
|
ns = str(meta.get("namespace") or "").strip()
|
|
item_name = str(meta.get("name") or "").strip()
|
|
if not ns or ns == target_namespace or item_name != name:
|
|
continue
|
|
_log(
|
|
log,
|
|
f"[SERVICES] Removing duplicate {kind}/{name} from namespace {ns} (target namespace: {target_namespace})",
|
|
)
|
|
_kubectl(
|
|
["-n", ns, "delete", kind, name, "--ignore-not-found"],
|
|
env=env,
|
|
timeout=60,
|
|
)
|
|
|
|
|
|
def _reconcile_deployment_replicasets(
|
|
*,
|
|
deployment: str,
|
|
namespace: str,
|
|
label_selector: str,
|
|
env: dict | None = None,
|
|
log: _LogFn | None = None,
|
|
) -> None:
|
|
dep_res = _kubectl(
|
|
["-n", namespace, "get", "deployment", deployment, "-o", "json"],
|
|
env=env,
|
|
timeout=30,
|
|
)
|
|
if dep_res.returncode != 0 or not dep_res.stdout.strip():
|
|
return
|
|
try:
|
|
dep = json.loads(dep_res.stdout)
|
|
except Exception:
|
|
return
|
|
|
|
dep_meta = dep.get("metadata") or {}
|
|
dep_uid = str(dep_meta.get("uid") or "").strip()
|
|
dep_annotations = dep_meta.get("annotations") or {}
|
|
try:
|
|
dep_revision = int(dep_annotations.get("deployment.kubernetes.io/revision") or "0")
|
|
except Exception:
|
|
dep_revision = 0
|
|
|
|
rs_res = _kubectl(
|
|
["-n", namespace, "get", "replicaset", "-l", label_selector, "-o", "json"],
|
|
env=env,
|
|
timeout=45,
|
|
)
|
|
if rs_res.returncode != 0 or not rs_res.stdout.strip():
|
|
return
|
|
try:
|
|
payload = json.loads(rs_res.stdout)
|
|
except Exception:
|
|
return
|
|
|
|
def _rs_revision(item: dict) -> int:
|
|
annotations = (item.get("metadata") or {}).get("annotations") or {}
|
|
try:
|
|
return int(annotations.get("deployment.kubernetes.io/revision") or "0")
|
|
except Exception:
|
|
return 0
|
|
|
|
owned: list[dict] = []
|
|
for item in payload.get("items", []):
|
|
meta = item.get("metadata") or {}
|
|
owners = meta.get("ownerReferences") or []
|
|
for owner in owners:
|
|
if str(owner.get("kind") or "") != "Deployment":
|
|
continue
|
|
if str(owner.get("name") or "") != deployment:
|
|
continue
|
|
owner_uid = str(owner.get("uid") or "").strip()
|
|
if dep_uid and owner_uid and owner_uid != dep_uid:
|
|
continue
|
|
owned.append(item)
|
|
break
|
|
|
|
active = [
|
|
item for item in owned if int((item.get("spec") or {}).get("replicas") or 0) > 0
|
|
]
|
|
if len(active) <= 1:
|
|
return
|
|
|
|
stale: list[dict]
|
|
if dep_revision > 0:
|
|
stale = [item for item in active if _rs_revision(item) < dep_revision]
|
|
else:
|
|
# Fallback: keep newest revision/creation and scale down the rest.
|
|
active_sorted = sorted(
|
|
active,
|
|
key=lambda item: (
|
|
_rs_revision(item),
|
|
str((item.get("metadata") or {}).get("creationTimestamp") or ""),
|
|
),
|
|
)
|
|
stale = active_sorted[:-1]
|
|
|
|
for item in stale:
|
|
rs_name = str((item.get("metadata") or {}).get("name") or "").strip()
|
|
if not rs_name:
|
|
continue
|
|
_log(
|
|
log,
|
|
f"[SERVICES] Scaling stale replicaset/{rs_name} to 0 in namespace {namespace} (deployment={deployment})",
|
|
)
|
|
_kubectl(
|
|
["-n", namespace, "scale", f"replicaset/{rs_name}", "--replicas=0"],
|
|
env=env,
|
|
timeout=60,
|
|
)
|