mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 12:03:59 +00:00
2f98a5afd2
install_barman_plugin() fetches from GitHub on every run (up to 6×120 s retries), stalling the installer worker thread even when the CNPG cluster is already at 'Cluster in healthy state' with all pods 2/2 Ready. Add a fast-path at the top of initialize(): check .status.phase for 'healthy' then verify all pods show 2/N ready — if both pass, return immediately. Full pipeline (ensure_operator, install_barman_plugin, _apply_manifest, _wait_cnpg_pods) is only entered when needed. Also adds test_cnpg_initialize_skips_when_cluster_healthy to test_deployment_mode_isolation.py to guard against regression. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2109 lines
78 KiB
Python
2109 lines
78 KiB
Python
"""
|
||
knoe/core/ops/cloudnative_pg.py
|
||
|
||
Canonical Python owner for CloudNativePG lifecycle behavior.
|
||
|
||
Replaces etc/init_cloudnative_pg.sh runtime dispatch for the four active
|
||
call paths:
|
||
|
||
initialize -- install operator, apply cluster manifests, wait for readiness
|
||
deploy -- image-aware manifest apply + instance reconcile
|
||
rollout -- rolling restart via ordered pod deletion / promotion
|
||
install_barman_plugin -- install the Barman Cloud operator plugin
|
||
|
||
Public API
|
||
----------
|
||
ensure_operator(env, log, mode) -> None
|
||
pin_controller(env, log, mode) -> None
|
||
install_barman_plugin(env, log) -> None
|
||
initialize(namespace, cluster_name, env, project_root, log, mode) -> None
|
||
deploy(namespace, cluster_name, env, project_root, log, mode) -> None
|
||
rollout(namespace, cluster_name, env, log) -> None
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import datetime
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import json
|
||
import logging
|
||
from cryptography import x509
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
from cryptography.hazmat.primitives.asymmetric.ec import (
|
||
SECP256R1,
|
||
generate_private_key,
|
||
)
|
||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||
from cryptography.x509.oid import NameOID
|
||
from typing import Callable
|
||
|
||
_LOG = logging.getLogger(__name__)
|
||
_LogFn = Callable[[str], None]
|
||
|
||
_CNPG_OPERATOR_FALLBACK_VERSION = "1.28.1"
|
||
_BARMAN_PLUGIN_FALLBACK_VERSION = "0.6.0"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Internal helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _log(log: _LogFn | None, msg: str) -> None:
|
||
if log is not None:
|
||
log(msg)
|
||
else:
|
||
_LOG.info(msg)
|
||
|
||
|
||
def _kubectl_cmd(args: list[str], env: dict | None = None) -> list[str]:
|
||
"""Build a kubectl command that honors env['KUBECONTEXT'] as --context."""
|
||
ctx = str((env or {}).get("KUBECONTEXT") or "").strip()
|
||
cmd = ["kubectl"]
|
||
if ctx:
|
||
cmd.extend(["--context", ctx])
|
||
cmd.extend(args)
|
||
return cmd
|
||
|
||
|
||
def _kubectl_run(
|
||
args: list[str],
|
||
env: dict | None = None,
|
||
timeout: int = 30,
|
||
check: bool = False,
|
||
input_text: str | None = None,
|
||
) -> subprocess.CompletedProcess:
|
||
"""Run kubectl with optional stdin payload and context-aware command building."""
|
||
return subprocess.run(
|
||
_kubectl_cmd(args, env=env),
|
||
env=env,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
check=check,
|
||
input=input_text,
|
||
)
|
||
|
||
|
||
def _kubectl(
|
||
args: list[str],
|
||
env: dict | None = None,
|
||
timeout: int = 30,
|
||
check: bool = False,
|
||
) -> subprocess.CompletedProcess:
|
||
return _kubectl_run(args, env=env, timeout=timeout, check=check)
|
||
|
||
|
||
def _kubectl_ok(args: list[str], env: dict | None = None, timeout: int = 15) -> bool:
|
||
return _kubectl(args, env=env, timeout=timeout).returncode == 0
|
||
|
||
|
||
def _release_stale_pv_claims(
|
||
storage_class: str = "synology-iscsi",
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Clear claimRef on Released PVs so they become Available for new PVCs.
|
||
|
||
Only touches PVs whose status.phase is 'Released' for the given storageClass.
|
||
Idempotent — Available/Bound PVs are left untouched.
|
||
"""
|
||
r = _kubectl(["get", "pv", "-o", "json"], env=env, timeout=20)
|
||
if r.returncode != 0:
|
||
_log(log, f"WARN: could not list PVs: {r.stderr.strip()}")
|
||
return
|
||
try:
|
||
pv_list = json.loads(r.stdout or "{}")
|
||
except Exception:
|
||
return
|
||
released = [
|
||
item["metadata"]["name"]
|
||
for item in (pv_list.get("items") or [])
|
||
if (item.get("spec", {}).get("storageClassName") == storage_class
|
||
and item.get("status", {}).get("phase") == "Released")
|
||
]
|
||
if not released:
|
||
return
|
||
_log(log, f"Releasing claimRef on {len(released)} Released '{storage_class}' PV(s): {', '.join(released)}")
|
||
for pv_name in released:
|
||
pr = _kubectl(
|
||
["patch", "pv", pv_name, "-p", '{"spec":{"claimRef":null}}'],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
if pr.returncode != 0:
|
||
_log(log, f"WARN: could not release claimRef on PV '{pv_name}': {pr.stderr.strip()}")
|
||
else:
|
||
_log(log, f"Released PV '{pv_name}'.")
|
||
|
||
|
||
def _reconcile_unbound_synology_data_wal_pvs(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Recreate unbound static Synology data/wal PVs from manifest.
|
||
|
||
This allows nodeAffinity/layout changes in `iscsi-pvs.yaml` to take effect,
|
||
while preserving currently bound volumes.
|
||
"""
|
||
pv_names = [
|
||
"synology-iscsi-d001-data",
|
||
"synology-iscsi-d001-wal",
|
||
"synology-iscsi-d002-data",
|
||
"synology-iscsi-d002-wal",
|
||
"synology-iscsi-d003-data",
|
||
"synology-iscsi-d003-wal",
|
||
]
|
||
|
||
unbound: list[str] = []
|
||
for name in pv_names:
|
||
r = _kubectl(["get", "pv", name, "-o", "json"], env=env, timeout=20)
|
||
if r.returncode != 0:
|
||
continue
|
||
try:
|
||
payload = json.loads(r.stdout or "{}")
|
||
except Exception:
|
||
continue
|
||
phase = str((payload.get("status") or {}).get("phase") or "").strip()
|
||
if phase != "Bound":
|
||
unbound.append(name)
|
||
|
||
if not unbound:
|
||
return
|
||
|
||
_log(log, f"Reconciling unbound Synology data/wal PVs: {', '.join(unbound)}")
|
||
for name in unbound:
|
||
_kubectl(["delete", "pv", name, "--ignore-not-found"], env=env, timeout=45)
|
||
|
||
root = Path(
|
||
(env or {}).get("KNOE_HOME")
|
||
or (env or {}).get("KNOE_SERVICE")
|
||
or Path(__file__).resolve().parents[4]
|
||
)
|
||
manifest = root / "k8s" / "knoe" / "iscsi-pvs.yaml"
|
||
if not manifest.exists():
|
||
_log(log, f"WARN: Synology PV manifest not found: {manifest}")
|
||
return
|
||
|
||
r_apply = _kubectl(["apply", "-f", str(manifest)], env=env, timeout=180)
|
||
if r_apply.returncode != 0:
|
||
stderr = ((r_apply.stderr or "") + "\n" + (r_apply.stdout or "")).strip().lower()
|
||
if "field is immutable" not in stderr:
|
||
raise RuntimeError(
|
||
f"Failed to re-apply Synology PV manifest {manifest}: "
|
||
f"{(r_apply.stderr or r_apply.stdout or '').strip()}"
|
||
)
|
||
_log(log, "WARN: Immutable PV fields prevented full re-apply; continuing with recreated unbound PVs.")
|
||
|
||
|
||
def _ensure_namespace(namespace: str, env: dict | None = None) -> None:
|
||
"""Create the namespace if it does not already exist (idempotent)."""
|
||
manifest = (
|
||
"apiVersion: v1\n"
|
||
"kind: Namespace\n"
|
||
"metadata:\n"
|
||
f" name: {namespace}\n"
|
||
)
|
||
res = _kubectl_run(["apply", "-f", "-"], env=env, input_text=manifest)
|
||
if res.returncode != 0:
|
||
raise RuntimeError(f"Failed to ensure namespace '{namespace}':\n{res.stderr}")
|
||
|
||
|
||
def _resolve_cnpg_manifest(project_root: str | Path, env: dict | None) -> Path:
|
||
"""Resolve the CNPG cluster manifest path, mirroring the shell K8S_PROLE_DIR logic."""
|
||
if env:
|
||
override = (env.get("CNPG_MANIFEST_OVERRIDE") or "").strip()
|
||
if override:
|
||
p = Path(override)
|
||
if p.exists():
|
||
return p
|
||
root = Path(project_root)
|
||
mode = str((env or {}).get("KNOE_MODE") or "").strip()
|
||
candidates: list[Path] = []
|
||
if mode == "k8s":
|
||
# GKE/prod: use the GCP-specific manifest (premium-rwo storage, no Synology selectors)
|
||
candidates.append(root / "deploy" / "gcp" / "gke" / "knoe-db.yaml")
|
||
candidates += [
|
||
root / "deploy" / "opentofu" / "k3s" / "manifests" / "knoe" / "knoe-db.yaml",
|
||
root / "k8s" / "knoe" / "knoe-db.yaml",
|
||
]
|
||
if env:
|
||
knoe_home = (env.get("KNOE_HOME") or "").strip()
|
||
if knoe_home:
|
||
ph = Path(knoe_home)
|
||
if mode == "k8s":
|
||
candidates.insert(0, ph / "deploy" / "gcp" / "gke" / "knoe-db.yaml")
|
||
candidates.append(ph / "k8s" / "knoe" / "knoe-db.yaml")
|
||
for candidate in candidates:
|
||
if candidate.exists():
|
||
return candidate
|
||
return root / "k8s" / "knoe" / "knoe-db.yaml"
|
||
|
||
|
||
def _resolve_knoe_db_image_tag(manifest: Path, env: dict | None) -> str:
|
||
explicit_tag = str((env or {}).get("KNOE_DB_IMAGE_TAG") or "").strip()
|
||
if explicit_tag:
|
||
return explicit_tag
|
||
explicit_tag = os.environ.get("KNOE_DB_IMAGE_TAG", "").strip()
|
||
if explicit_tag:
|
||
return explicit_tag
|
||
|
||
mode_key = str((env or {}).get("KNOE_MODE") or "").strip() or "k8s"
|
||
|
||
candidate_roots: list[Path] = []
|
||
knoe_home = str((env or {}).get("KNOE_HOME") or "").strip()
|
||
if knoe_home:
|
||
candidate_roots.append(Path(knoe_home))
|
||
|
||
for parent in [manifest, *manifest.parents]:
|
||
if parent in candidate_roots:
|
||
continue
|
||
if (parent / "conf").exists() or (parent / "knoe-db").exists() or (parent / "modes").exists():
|
||
candidate_roots.append(parent)
|
||
|
||
def _read(path: Path) -> str | None:
|
||
try:
|
||
val = path.read_text(encoding="utf-8").strip()
|
||
return val or None
|
||
except OSError:
|
||
return None
|
||
|
||
pg_version = "17.7"
|
||
release = "43"
|
||
for root in candidate_roots:
|
||
mode_pg = root / "modes" / mode_key / "conf" / "postgresql" / ".version"
|
||
mode_release = root / "modes" / mode_key / "knoe-db" / ".version"
|
||
root_pg = root / "conf" / "postgresql" / ".version"
|
||
root_release = root / "knoe-db" / ".version"
|
||
|
||
if mode_pg.exists() or mode_release.exists() or root_pg.exists() or root_release.exists():
|
||
pg_version = _read(mode_pg) or _read(root_pg) or pg_version
|
||
release = _read(mode_release) or _read(root_release) or release
|
||
break
|
||
|
||
if release.isdigit():
|
||
release = release.zfill(3)
|
||
return f"{pg_version}-{release}"
|
||
|
||
|
||
def _strip_k3d_synology_blocks(text: str) -> str:
|
||
lines = text.splitlines()
|
||
out: list[str] = []
|
||
skip_block = False
|
||
skip_indent = 0
|
||
i = 0
|
||
while i < len(lines):
|
||
line = lines[i]
|
||
stripped = line.lstrip()
|
||
if skip_block:
|
||
if not stripped:
|
||
i += 1
|
||
continue
|
||
indent = len(line) - len(stripped)
|
||
if indent > skip_indent:
|
||
i += 1
|
||
continue
|
||
skip_block = False
|
||
skip_indent = 0
|
||
continue
|
||
|
||
if stripped.startswith("affinity:"):
|
||
block_indent = len(line) - len(stripped)
|
||
block_lines = [line]
|
||
j = i + 1
|
||
while j < len(lines):
|
||
next_line = lines[j]
|
||
next_stripped = next_line.lstrip()
|
||
if not next_stripped:
|
||
block_lines.append(next_line)
|
||
j += 1
|
||
continue
|
||
next_indent = len(next_line) - len(next_stripped)
|
||
if next_indent <= block_indent:
|
||
break
|
||
block_lines.append(next_line)
|
||
j += 1
|
||
if any("kubernetes.io/hostname" in bl for bl in block_lines):
|
||
i = j
|
||
continue
|
||
out.extend(block_lines)
|
||
i = j
|
||
continue
|
||
|
||
if stripped.startswith("selector:"):
|
||
block_indent = len(line) - len(stripped)
|
||
block_lines = [line]
|
||
j = i + 1
|
||
while j < len(lines):
|
||
next_line = lines[j]
|
||
next_stripped = next_line.lstrip()
|
||
if not next_stripped:
|
||
block_lines.append(next_line)
|
||
j += 1
|
||
continue
|
||
next_indent = len(next_line) - len(next_stripped)
|
||
if next_indent <= block_indent:
|
||
break
|
||
block_lines.append(next_line)
|
||
j += 1
|
||
if any("synology.storage/" in bl for bl in block_lines):
|
||
i = j
|
||
continue
|
||
out.extend(block_lines)
|
||
i = j
|
||
continue
|
||
|
||
if stripped.startswith("nodeSelector:"):
|
||
skip_block = True
|
||
skip_indent = len(line) - len(stripped)
|
||
i += 1
|
||
continue
|
||
|
||
if stripped.startswith("storageClassName:") and "synology-iscsi" in stripped:
|
||
i += 1
|
||
continue
|
||
|
||
out.append(line)
|
||
i += 1
|
||
|
||
tail = "\n" if text.endswith("\n") else ""
|
||
return "\n".join(out) + tail
|
||
|
||
|
||
def _apply_manifest(
|
||
namespace: str,
|
||
manifest: Path,
|
||
env: dict | None,
|
||
log: _LogFn | None,
|
||
attempts: int = 8,
|
||
) -> None:
|
||
"""Apply a CNPG cluster manifest with retry on transient API errors."""
|
||
if not manifest.exists():
|
||
raise FileNotFoundError(f"CNPG manifest not found: {manifest}")
|
||
|
||
raw_manifest = manifest.read_text(encoding="utf-8")
|
||
rendered_manifest = raw_manifest
|
||
artifact_registry = ""
|
||
if "${ARTIFACT_REGISTRY}" in rendered_manifest:
|
||
artifact_registry = str((env or {}).get("ARTIFACT_REGISTRY") or "").strip()
|
||
if not artifact_registry:
|
||
artifact_registry = os.environ.get("ARTIFACT_REGISTRY", "").strip()
|
||
if not artifact_registry:
|
||
raise RuntimeError(
|
||
"CNPG manifest contains '${ARTIFACT_REGISTRY}' but ARTIFACT_REGISTRY is not set."
|
||
)
|
||
rendered_manifest = rendered_manifest.replace("${ARTIFACT_REGISTRY}", artifact_registry.rstrip("/"))
|
||
|
||
if "${KNOE_DB_IMAGE_TAG}" in rendered_manifest:
|
||
knoe_db_image_tag = _resolve_knoe_db_image_tag(manifest, env)
|
||
rendered_manifest = rendered_manifest.replace("${KNOE_DB_IMAGE_TAG}", knoe_db_image_tag)
|
||
|
||
if str((env or {}).get("KNOE_MODE") or "").strip().lower() == "k3d":
|
||
rendered_manifest = _strip_k3d_synology_blocks(rendered_manifest)
|
||
|
||
# Resolve ${KNOE_IMAGE_REGISTRY} — the in-cluster container registry hostname.
|
||
# Resolution order:
|
||
# 1. KNOE_IMAGE_REGISTRY env/config (explicit wins)
|
||
# 2. k3d mode fallback: k3d-<registry_name>:<port>
|
||
# 3. If GitLab is deployed: gitlab-registry.<gitlab_ns>.svc.cluster.local:5000
|
||
# 4. Fallback: registry.<service_ns>.svc.cluster.local:5000 (registry:2)
|
||
if "${KNOE_IMAGE_REGISTRY}" in rendered_manifest:
|
||
knoe_image_registry = str((env or {}).get("KNOE_IMAGE_REGISTRY") or "").strip()
|
||
if not knoe_image_registry:
|
||
knoe_image_registry = os.environ.get("KNOE_IMAGE_REGISTRY", "").strip()
|
||
if not knoe_image_registry:
|
||
knoe_mode = str((env or {}).get("KNOE_MODE") or "").strip().lower()
|
||
if knoe_mode == "k3d":
|
||
registry_name = str((env or {}).get("K3D_REGISTRY_NAME") or "knoe-registry").strip() or "knoe-registry"
|
||
registry_port = str((env or {}).get("REGISTRY_PORT") or "5000").strip() or "5000"
|
||
registry_host = registry_name if registry_name.startswith("k3d-") else f"k3d-{registry_name}"
|
||
knoe_image_registry = f"{registry_host}:{registry_port}"
|
||
_log(log, f"INFO: Using k3d registry as KNOE_IMAGE_REGISTRY ({knoe_image_registry})")
|
||
elif knoe_mode == "k3s":
|
||
# For k3s, we MUST NOT use .svc.cluster.local because nodes need to resolve it.
|
||
# Default to registry.knoe.org if not explicitly set.
|
||
knoe_image_registry = "registry.knoe.org"
|
||
_log(log, f"INFO: Defaulting to registry.knoe.org for KNOE_IMAGE_REGISTRY in k3s mode.")
|
||
|
||
if not knoe_image_registry:
|
||
# Auto-detect (non-k3s path)
|
||
gitlab_ns = str((env or {}).get("GITLAB_NAMESPACE") or "gitlab").strip()
|
||
service_ns = str((env or {}).get("SERVICE_NAMESPACE") or "knoe-system").strip()
|
||
try:
|
||
r = _kubectl(["get", "namespace", gitlab_ns], env=env, timeout=10)
|
||
if r.returncode == 0:
|
||
knoe_image_registry = f"gitlab-registry.{gitlab_ns}.svc.cluster.local:5000"
|
||
_log(log, f"INFO: GitLab detected — using gitlab-registry as KNOE_IMAGE_REGISTRY ({knoe_image_registry})")
|
||
except Exception:
|
||
pass
|
||
if not knoe_image_registry:
|
||
service_ns = str((env or {}).get("SERVICE_NAMESPACE") or "knoe-system").strip()
|
||
knoe_image_registry = f"registry.{service_ns}.svc.cluster.local:5000"
|
||
_log(log, f"INFO: Using registry:2 as KNOE_IMAGE_REGISTRY ({knoe_image_registry})")
|
||
rendered_manifest = rendered_manifest.replace(
|
||
"${KNOE_IMAGE_REGISTRY}", knoe_image_registry.rstrip("/")
|
||
)
|
||
|
||
# Per-cluster image override: CNPG_IMAGE_NAME replaces the imageName line entirely.
|
||
# Useful when different clusters use different PostgreSQL builds or versions.
|
||
cnpg_image_name = str((env or {}).get("CNPG_IMAGE_NAME") or "").strip()
|
||
if cnpg_image_name:
|
||
rendered_manifest = re.sub(
|
||
r"^(\s*imageName:\s*).*$",
|
||
rf"\g<1>{cnpg_image_name}",
|
||
rendered_manifest,
|
||
flags=re.MULTILINE,
|
||
)
|
||
|
||
unresolved_image_placeholder = re.search(
|
||
r"^\s*imageName:\s*[\"']?\$\{[^}]+}[^\s\"']*",
|
||
rendered_manifest,
|
||
flags=re.MULTILINE,
|
||
)
|
||
if unresolved_image_placeholder:
|
||
raise RuntimeError(
|
||
"CNPG manifest contains unresolved image placeholder in imageName. "
|
||
"Ensure deployment variables are rendered before apply."
|
||
)
|
||
|
||
manifest_path = manifest
|
||
temp_manifest_path: Path | None = None
|
||
if rendered_manifest != raw_manifest:
|
||
with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".yaml", delete=False) as tmp:
|
||
tmp.write(rendered_manifest)
|
||
temp_manifest_path = Path(tmp.name)
|
||
manifest_path = temp_manifest_path
|
||
|
||
_log(log, f"Applying CNPG manifest {manifest} in namespace {namespace}...")
|
||
last_err = ""
|
||
try:
|
||
for i in range(1, attempts + 1):
|
||
r = _kubectl(["-n", namespace, "apply", "-f", str(manifest_path)], env=env, timeout=60)
|
||
if r.returncode == 0:
|
||
if r.stdout.strip():
|
||
_log(log, r.stdout.strip())
|
||
return
|
||
last_err = (r.stderr or r.stdout or "").strip()
|
||
transient = any(
|
||
kw in last_err.lower()
|
||
for kw in ("etcdserver", "connection refused", "timeout", "tls handshake", "webhook")
|
||
)
|
||
if transient and i < attempts:
|
||
_log(log, f"WARN: manifest apply failed (attempt {i}/{attempts}); retrying in 5s...")
|
||
time.sleep(5)
|
||
continue
|
||
break
|
||
raise RuntimeError(f"CNPG manifest apply failed after {attempts} attempts: {last_err}")
|
||
finally:
|
||
if temp_manifest_path is not None:
|
||
try:
|
||
temp_manifest_path.unlink()
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _remove_legacy_barman_object_store_when_plugin_is_enabled(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None,
|
||
log: _LogFn | None,
|
||
) -> None:
|
||
"""Avoid CNPG validation conflict when plugin mode coexists with legacy barmanObjectStore."""
|
||
exists = _kubectl(["-n", namespace, "get", "cluster", cluster_name, "-o", "name"], env=env, timeout=20)
|
||
if exists.returncode != 0:
|
||
return
|
||
|
||
plugins = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={range .spec.plugins[*]}{.name}{\n}{end}"],
|
||
env=env,
|
||
timeout=20,
|
||
)
|
||
if plugins.returncode != 0:
|
||
return
|
||
plugin_names = {line.strip() for line in (plugins.stdout or "").splitlines() if line.strip()}
|
||
plugin_enabled = "barman-cloud.cloudnative-pg.io" in plugin_names
|
||
if not plugin_enabled:
|
||
return
|
||
|
||
destination = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.spec.backup.barmanObjectStore.destinationPath}"],
|
||
env=env,
|
||
timeout=20,
|
||
)
|
||
if destination.returncode != 0 or not (destination.stdout or "").strip():
|
||
return
|
||
|
||
_log(log, "Removing legacy spec.backup.barmanObjectStore to keep plugin WAL archiver valid...")
|
||
patch = json.dumps({"spec": {"backup": {"barmanObjectStore": None}}})
|
||
patched = _kubectl(
|
||
["-n", namespace, "patch", "cluster", cluster_name, "--type", "merge", "-p", patch],
|
||
env=env,
|
||
timeout=30,
|
||
)
|
||
if patched.returncode != 0:
|
||
reason = (patched.stderr or patched.stdout or "").strip()
|
||
raise RuntimeError(
|
||
f"Failed to clear legacy barmanObjectStore on cluster '{cluster_name}' in namespace '{namespace}': {reason}"
|
||
)
|
||
|
||
|
||
def _wait_pod_ready(
|
||
namespace: str, pod: str, env: dict | None, timeout: int = 300, log: _LogFn | None = None
|
||
) -> None:
|
||
r = _kubectl(
|
||
["-n", namespace, "wait", "--for=condition=Ready", f"pod/{pod}", f"--timeout={timeout}s"],
|
||
env=env,
|
||
timeout=timeout + 10,
|
||
)
|
||
if r.returncode != 0:
|
||
_log(log, f"WARN: pod {pod} did not become Ready within {timeout}s")
|
||
|
||
|
||
def _cnpg_pod_readiness_lines(namespace: str, cluster_name: str, env: dict | None) -> list[str]:
|
||
r = _kubectl(
|
||
[
|
||
"-n", namespace, "get", "pods",
|
||
"-l", f"cnpg.io/cluster={cluster_name}",
|
||
"-o",
|
||
"jsonpath={range .items[*]}{.metadata.name}={.status.conditions[?(@.type==\"Ready\")].status}\\n{end}",
|
||
],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
if r.returncode != 0:
|
||
return []
|
||
# kubectl jsonpath emits literal \n (backslash-n) not real newlines
|
||
raw = (r.stdout or "").replace("\\n", "\n")
|
||
return [line for line in raw.splitlines() if "=" in line]
|
||
|
||
|
||
def _cnpg_timeout_diagnostics(namespace: str, cluster_name: str, env: dict | None) -> str:
|
||
parts: list[str] = []
|
||
|
||
pod_status = _kubectl(
|
||
[
|
||
"-n", namespace, "get", "pods",
|
||
"-l", f"cnpg.io/cluster={cluster_name}",
|
||
"-o",
|
||
"jsonpath={range .items[*]}{.metadata.name}|phase={.status.phase}|node={.spec.nodeName}|ready={.status.conditions[?(@.type==\"Ready\")].status}|reason={.status.containerStatuses[0].state.waiting.reason}\\n{end}",
|
||
],
|
||
env=env,
|
||
timeout=20,
|
||
)
|
||
if pod_status.returncode == 0:
|
||
pod_lines = [line for line in (pod_status.stdout or "").replace("\\n", "\n").splitlines() if line.strip()]
|
||
if pod_lines:
|
||
parts.append("Pods: " + "; ".join(pod_lines))
|
||
|
||
failed_sched = _kubectl(
|
||
[
|
||
"-n", namespace, "get", "events",
|
||
"--field-selector", "reason=FailedScheduling,involvedObject.kind=Pod",
|
||
"-o", "jsonpath={range .items[*]}{.involvedObject.name}|{.message}\\n{end}",
|
||
],
|
||
env=env,
|
||
timeout=20,
|
||
)
|
||
if failed_sched.returncode == 0:
|
||
event_lines = [line for line in (failed_sched.stdout or "").replace("\\n", "\n").splitlines() if line.strip()]
|
||
if event_lines:
|
||
parts.append("FailedScheduling: " + " | ".join(event_lines[-3:]))
|
||
|
||
if not parts:
|
||
return "No additional scheduling diagnostics available."
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _relax_cnpg_workload_node_selector_if_unmatched(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None,
|
||
log: _LogFn | None,
|
||
) -> bool:
|
||
cfg = env or {}
|
||
relax_enabled = str(cfg.get("CNPG_AUTO_RELAX_WORKLOAD_SELECTOR", "true") or "true").strip().lower()
|
||
if relax_enabled in {"0", "false", "no", "off"}:
|
||
return False
|
||
|
||
selector_key = str(cfg.get("CNPG_WORKLOAD_SELECTOR_KEY", "workload") or "workload").strip() or "workload"
|
||
selector_value_res = _kubectl(
|
||
[
|
||
"-n", namespace, "get", "cluster", cluster_name,
|
||
"-o", f"jsonpath={{.spec.affinity.nodeSelector.{selector_key}}}",
|
||
],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
selector_value = (selector_value_res.stdout or "").strip()
|
||
if selector_value_res.returncode != 0 or not selector_value:
|
||
return False
|
||
|
||
selector_expr = f"{selector_key}={selector_value}"
|
||
matching_nodes = _kubectl(["get", "nodes", "-l", selector_expr, "--no-headers"], env=env, timeout=15)
|
||
if matching_nodes.returncode != 0:
|
||
return False
|
||
matched = len([line for line in (matching_nodes.stdout or "").splitlines() if line.strip()])
|
||
if matched > 0:
|
||
return False
|
||
|
||
_log(
|
||
log,
|
||
"WARN: CNPG pods appear unschedulable because no nodes match "
|
||
f"'{selector_expr}'. Removing spec.affinity.nodeSelector as tiny-cluster fallback.",
|
||
)
|
||
patch = json.dumps({"spec": {"affinity": {"nodeSelector": None}}})
|
||
patched = _kubectl(
|
||
["-n", namespace, "patch", "cluster", cluster_name, "--type", "merge", "-p", patch],
|
||
env=env,
|
||
timeout=30,
|
||
)
|
||
if patched.returncode != 0:
|
||
reason = (patched.stderr or patched.stdout or "").strip()
|
||
_log(log, f"WARN: failed to relax CNPG nodeSelector: {reason}")
|
||
return False
|
||
|
||
_log(log, "CNPG cluster nodeSelector relaxed successfully; retrying pod readiness wait.")
|
||
return True
|
||
|
||
|
||
def _wait_cnpg_pods(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None,
|
||
timeout: int = 900,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Wait for all CNPG cluster pods to reach Ready state."""
|
||
# Determine expected pod count from cluster spec
|
||
r = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.spec.instances}"],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
target = 1
|
||
if r.returncode == 0:
|
||
try:
|
||
target = max(1, int((r.stdout or "1").strip() or "1"))
|
||
except ValueError:
|
||
target = 1
|
||
|
||
_log(log, f"Waiting for {target} CNPG pod(s) for cluster '{cluster_name}' in '{namespace}' (timeout={timeout}s)...")
|
||
|
||
deadline = time.monotonic() + timeout
|
||
selector_relaxed = False
|
||
last_feedback = 0.0
|
||
while True:
|
||
now = time.monotonic()
|
||
if now >= deadline:
|
||
if not selector_relaxed and _relax_cnpg_workload_node_selector_if_unmatched(namespace, cluster_name, env, log):
|
||
selector_relaxed = True
|
||
grace = int((env or {}).get("CNPG_SELECTOR_RELAX_GRACE_TIMEOUT", "300") or "300")
|
||
timeout = max(30, grace)
|
||
deadline = now + timeout
|
||
last_feedback = 0.0
|
||
_log(log, f"Waiting up to {timeout}s for CNPG pods after nodeSelector relaxation...")
|
||
continue
|
||
|
||
diagnostics = _cnpg_timeout_diagnostics(namespace, cluster_name, env)
|
||
raise RuntimeError(
|
||
f"Timed out ({timeout}s) waiting for {target} CNPG pods to be Ready "
|
||
f"in namespace '{namespace}'.\n"
|
||
f"{diagnostics}"
|
||
)
|
||
|
||
lines = _cnpg_pod_readiness_lines(namespace, cluster_name, env)
|
||
if lines:
|
||
ready = sum(1 for l in lines if l.split("=", 1)[1].strip() == "True")
|
||
if ready >= target:
|
||
_log(log, f"All {ready}/{target} CNPG pods are Ready.")
|
||
return
|
||
|
||
elapsed = timeout - (deadline - now)
|
||
if elapsed - last_feedback >= 30:
|
||
last_feedback = elapsed
|
||
_log(log, f" [{int(elapsed)}s/{timeout}s] {ready}/{target} CNPG pods ready...")
|
||
|
||
time.sleep(5)
|
||
|
||
|
||
def _reconcile_instances(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None,
|
||
log: _LogFn | None,
|
||
) -> None:
|
||
"""
|
||
Cap CNPG cluster instances to the number of eligible schedulable nodes,
|
||
then patch the cluster spec if the count has changed.
|
||
|
||
Mirrors reconcile_cnpg_instances() from the shell script.
|
||
"""
|
||
# Current spec instances
|
||
r = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.spec.instances}"],
|
||
env=env, timeout=15,
|
||
)
|
||
if r.returncode != 0 or not (r.stdout or "").strip():
|
||
return # cluster may not exist yet
|
||
try:
|
||
current = max(0, int((r.stdout or "0").strip()))
|
||
except ValueError:
|
||
return
|
||
|
||
desired_raw = int((env or {}).get("CNPG_INSTANCES", "3") or "3")
|
||
desired = max(1, desired_raw)
|
||
|
||
# Count eligible nodes (nodes with cnpg.io/node-role=db or all schedulable nodes)
|
||
node_selector = (env or {}).get("CNPG_NODE_SELECTOR", "")
|
||
if node_selector:
|
||
r2 = _kubectl(["get", "nodes", "-l", node_selector, "--no-headers"], env=env, timeout=10)
|
||
else:
|
||
r2 = _kubectl(["get", "nodes", "--no-headers"], env=env, timeout=10)
|
||
eligible = len([l for l in (r2.stdout or "").splitlines() if l.strip()]) if r2.returncode == 0 else 1
|
||
eligible = max(1, eligible)
|
||
|
||
if desired > eligible:
|
||
desired = eligible
|
||
|
||
if current == desired:
|
||
return
|
||
|
||
# Clamp maxSyncReplicas so the spec stays valid (maxSyncReplicas < instances)
|
||
base_max_sync = int((env or {}).get("CNPG_MAX_SYNC_REPLICAS", "1") or "1")
|
||
if desired <= 1:
|
||
max_sync = 0
|
||
else:
|
||
max_sync = min(base_max_sync, desired - 1)
|
||
|
||
_log(log, f"Reconciling CNPG instances: {current} → {desired} (eligible_nodes={eligible})")
|
||
patch = json.dumps({"spec": {"instances": desired, "maxSyncReplicas": max_sync}})
|
||
for attempt in range(1, 9):
|
||
r3 = _kubectl(
|
||
["-n", namespace, "patch", "cluster", cluster_name, "--type", "merge", "-p", patch],
|
||
env=env, timeout=30,
|
||
)
|
||
if r3.returncode == 0:
|
||
_log(log, f"Cluster instances patched to {desired}.")
|
||
return
|
||
err = (r3.stderr or r3.stdout or "").strip().lower()
|
||
transient = any(kw in err for kw in ("etcd", "connection refused", "timeout", "webhook"))
|
||
if transient and attempt < 8:
|
||
_log(log, f"WARN: patch failed (attempt {attempt}/8); retrying in 5s...")
|
||
time.sleep(5)
|
||
else:
|
||
_log(log, f"WARN: could not patch CNPG instances: {err}")
|
||
return
|
||
|
||
|
||
def _controller_manifest_url(version: str) -> str:
|
||
minor = ".".join(version.split(".")[:2])
|
||
return (
|
||
f"https://raw.githubusercontent.com/cloudnative-pg/cloudnative-pg"
|
||
f"/release-{minor}/releases/cnpg-{version}.yaml"
|
||
)
|
||
|
||
|
||
def _controller_deployment_json(env: dict | None = None) -> dict | None:
|
||
r = _kubectl(
|
||
["-n", "cnpg-system", "get", "deploy", "cnpg-controller-manager", "-o", "json"],
|
||
env=env,
|
||
timeout=30,
|
||
)
|
||
if r.returncode != 0 or not (r.stdout or "").strip():
|
||
return None
|
||
try:
|
||
return json.loads(r.stdout)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _controller_is_healthy(env: dict | None = None) -> bool:
|
||
dep = _controller_deployment_json(env)
|
||
if not dep:
|
||
return False
|
||
spec = dep.get("spec") or {}
|
||
status = dep.get("status") or {}
|
||
desired = int(spec.get("replicas") or 1)
|
||
available = int(status.get("availableReplicas") or 0)
|
||
return desired > 0 and available >= desired
|
||
|
||
|
||
def _reconcile_controller_replicasets(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
aggressive_cleanup: bool = False,
|
||
) -> None:
|
||
dep = _controller_deployment_json(env)
|
||
if not dep:
|
||
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
|
||
|
||
r = _kubectl(
|
||
[
|
||
"-n",
|
||
"cnpg-system",
|
||
"get",
|
||
"rs",
|
||
"-l",
|
||
"app.kubernetes.io/name=cloudnative-pg",
|
||
"-o",
|
||
"json",
|
||
],
|
||
env=env,
|
||
timeout=45,
|
||
)
|
||
if r.returncode != 0 or not (r.stdout or "").strip():
|
||
return
|
||
try:
|
||
payload = json.loads(r.stdout)
|
||
except Exception:
|
||
return
|
||
|
||
def _rs_revision(item: dict) -> int:
|
||
ann = (item.get("metadata") or {}).get("annotations") or {}
|
||
try:
|
||
return int(ann.get("deployment.kubernetes.io/revision") or "0")
|
||
except Exception:
|
||
return 0
|
||
|
||
owned: list[dict] = []
|
||
for item in payload.get("items") or []:
|
||
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 "") != "cnpg-controller-manager":
|
||
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 = [i for i in owned if int((i.get("spec") or {}).get("replicas") or 0) > 0]
|
||
if len(active) <= 1:
|
||
return
|
||
|
||
stale: list[dict]
|
||
if dep_revision > 0:
|
||
stale = [i for i in active if _rs_revision(i) < dep_revision]
|
||
else:
|
||
active_sorted = sorted(
|
||
active,
|
||
key=lambda i: (
|
||
_rs_revision(i),
|
||
str((i.get("metadata") or {}).get("creationTimestamp") or ""),
|
||
),
|
||
)
|
||
stale = active_sorted[:-1]
|
||
|
||
for rs in stale:
|
||
rs_meta = rs.get("metadata") or {}
|
||
rs_name = str(rs_meta.get("name") or "").strip()
|
||
if not rs_name:
|
||
continue
|
||
_log(log, f"Scaling stale CNPG controller replicaset/{rs_name} to 0")
|
||
_kubectl(
|
||
["-n", "cnpg-system", "scale", f"rs/{rs_name}", "--replicas=0"],
|
||
env=env,
|
||
timeout=60,
|
||
)
|
||
|
||
if aggressive_cleanup:
|
||
template_hash = str((rs_meta.get("labels") or {}).get("pod-template-hash") or "").strip()
|
||
if template_hash:
|
||
_kubectl(
|
||
[
|
||
"-n",
|
||
"cnpg-system",
|
||
"delete",
|
||
"pod",
|
||
"-l",
|
||
f"app.kubernetes.io/name=cloudnative-pg,pod-template-hash={template_hash}",
|
||
"--ignore-not-found",
|
||
"--wait=false",
|
||
],
|
||
env=env,
|
||
timeout=60,
|
||
)
|
||
|
||
|
||
def _wait_controller_rollout(
|
||
env: dict | None = None,
|
||
timeout_seconds: int = 180,
|
||
) -> bool:
|
||
r = _kubectl(
|
||
[
|
||
"-n",
|
||
"cnpg-system",
|
||
"rollout",
|
||
"status",
|
||
"deploy/cnpg-controller-manager",
|
||
f"--timeout={timeout_seconds}s",
|
||
],
|
||
env=env,
|
||
timeout=timeout_seconds + 10,
|
||
)
|
||
return r.returncode == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Public API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def ensure_operator(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
mode: str = "",
|
||
allow_reapply_if_unhealthy: bool = False,
|
||
aggressive_cleanup: bool = False,
|
||
) -> None:
|
||
"""Install the CloudNativePG operator if not already present, then wait for readiness."""
|
||
if _kubectl_ok(["-n", "cnpg-system", "get", "deploy", "cnpg-controller-manager"], env=env):
|
||
_reconcile_controller_replicasets(env=env, log=log, aggressive_cleanup=aggressive_cleanup)
|
||
if _controller_is_healthy(env=env):
|
||
_log(log, "CloudNative-PG operator already healthy; skipping operator re-apply.")
|
||
if mode == "k3s":
|
||
_tune_operator_for_k3s(env, log)
|
||
_wait_controller_rollout(env=env, timeout_seconds=300)
|
||
return
|
||
|
||
_log(log, "CloudNative-PG operator exists but is not healthy; waiting for recovery...")
|
||
if _wait_controller_rollout(env=env, timeout_seconds=180):
|
||
if mode == "k3s":
|
||
_tune_operator_for_k3s(env, log)
|
||
_wait_controller_rollout(env=env, timeout_seconds=300)
|
||
return
|
||
|
||
if not allow_reapply_if_unhealthy:
|
||
raise RuntimeError(
|
||
"CNPG operator deployment exists but is not ready; "
|
||
"refusing disruptive re-apply in non-initialize flow."
|
||
)
|
||
_log(log, "CNPG operator unhealthy after wait; attempting controlled re-apply...")
|
||
|
||
version = ""
|
||
if env:
|
||
version = (env.get("CNPG_OPERATOR_VERSION") or env.get("CNPG_VERSION") or "").strip()
|
||
if not version or version == "latest":
|
||
version = _CNPG_OPERATOR_FALLBACK_VERSION
|
||
url = _controller_manifest_url(version)
|
||
_log(log, f"Installing CloudNative-PG operator version {version}...")
|
||
r = _kubectl(["apply", "--server-side", "-f", url], env=env, timeout=120)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(f"Failed to install CNPG operator: {(r.stderr or r.stdout or '').strip()}")
|
||
|
||
if _kubectl_ok(["-n", "cnpg-system", "get", "deploy", "cnpg-controller-manager"], env=env, timeout=20):
|
||
_reconcile_controller_replicasets(env=env, log=log, aggressive_cleanup=aggressive_cleanup)
|
||
if not _wait_controller_rollout(env=env, timeout_seconds=180):
|
||
raise RuntimeError("CNPG controller rollout did not complete after operator apply")
|
||
if mode == "k3s":
|
||
_tune_operator_for_k3s(env, log)
|
||
if not _wait_controller_rollout(env=env, timeout_seconds=300):
|
||
raise RuntimeError("CNPG controller rollout did not complete after k3s tuning")
|
||
|
||
|
||
def _tune_operator_for_k3s(env: dict | None, log: _LogFn | None) -> None:
|
||
"""Relax CNPG controller probes/resources for single-node k3s environments."""
|
||
_log(log, "Tuning CNPG operator deployment probes/resources for k3s...")
|
||
patch = json.dumps({
|
||
"spec": {
|
||
"template": {
|
||
"spec": {
|
||
"tolerations": [
|
||
{
|
||
"key": "node-role.kubernetes.io/control-plane",
|
||
"operator": "Exists",
|
||
"effect": "NoSchedule",
|
||
},
|
||
{
|
||
"key": "node-role.kubernetes.io/master",
|
||
"operator": "Exists",
|
||
"effect": "NoSchedule",
|
||
},
|
||
],
|
||
"containers": [
|
||
{
|
||
"name": "manager",
|
||
"resources": {
|
||
"requests": {"cpu": "250m", "memory": "512Mi"},
|
||
"limits": {"cpu": "500m", "memory": "1Gi"},
|
||
},
|
||
"livenessProbe": {"timeoutSeconds": 5, "failureThreshold": 6},
|
||
"readinessProbe": {"timeoutSeconds": 5, "failureThreshold": 6},
|
||
"startupProbe": {"timeoutSeconds": 5, "failureThreshold": 60},
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
})
|
||
_kubectl(
|
||
["-n", "cnpg-system", "patch", "deploy", "cnpg-controller-manager", "--type", "merge", "-p", patch],
|
||
env=env, timeout=30,
|
||
)
|
||
|
||
|
||
def _clear_controller_pin(env: dict | None = None, log: _LogFn | None = None) -> None:
|
||
"""Remove stale hard nodeSelector pinning for controller-manager."""
|
||
_log(log, "Clearing CNPG controller hard nodeSelector pin (if present)...")
|
||
patch = json.dumps({"spec": {"template": {"spec": {"nodeSelector": None}}}})
|
||
_kubectl(
|
||
["-n", "cnpg-system", "patch", "deployment", "cnpg-controller-manager", "--type", "merge", "-p", patch],
|
||
env=env,
|
||
timeout=30,
|
||
)
|
||
|
||
|
||
def pin_controller(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
mode: str = "",
|
||
) -> None:
|
||
"""Pin the CNPG controller manager to storage-designated nodes (k3s) or as configured."""
|
||
selector = (env or {}).get("CNPG_CONTROLLER_NODE_SELECTOR", "").strip()
|
||
|
||
if not selector and mode == "k3s":
|
||
r = _kubectl(["get", "nodes", "-l", "storage=primary", "--no-headers"], env=env, timeout=10)
|
||
if r.returncode == 0 and (r.stdout or "").strip():
|
||
selector = "storage=primary"
|
||
|
||
if not selector:
|
||
_clear_controller_pin(env=env, log=log)
|
||
_reconcile_controller_replicasets(env=env, log=log)
|
||
return
|
||
|
||
if "=" not in selector:
|
||
_log(log, f"WARN: CNPG_CONTROLLER_NODE_SELECTOR must be key=value (got '{selector}'). Skipping pin.")
|
||
return
|
||
|
||
key, value = selector.split("=", 1)
|
||
if not _kubectl_ok(["get", "nodes", "-l", f"{key}={value}", "--no-headers"], env=env):
|
||
_log(log, f"WARN: No nodes match CNPG_CONTROLLER_NODE_SELECTOR={selector}. Skipping pin.")
|
||
return
|
||
|
||
_log(log, f"Pinning cnpg-controller-manager to nodes with {selector}...")
|
||
patch = json.dumps({"spec": {"template": {"spec": {"nodeSelector": {key: value}}}}})
|
||
_kubectl(
|
||
["-n", "cnpg-system", "patch", "deployment", "cnpg-controller-manager",
|
||
"--type", "merge", "-p", patch],
|
||
env=env, timeout=30,
|
||
)
|
||
_reconcile_controller_replicasets(env=env, log=log)
|
||
|
||
|
||
def install_barman_plugin(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Install the Barman Cloud operator plugin, retrying on cert-manager webhook errors."""
|
||
url = ((env or {}).get("BARMAN_PLUGIN_MANIFEST_URL") or "").strip()
|
||
if not url:
|
||
tag = _resolve_barman_plugin_tag(env, log)
|
||
url = f"https://github.com/cloudnative-pg/plugin-barman-cloud/releases/download/{tag}/manifest.yaml"
|
||
|
||
_log(log, f"Installing Barman Cloud plugin from {url}...")
|
||
_CERT_WEBHOOK_PATTERNS = (
|
||
"webhook.cert-manager.io",
|
||
"cert-manager",
|
||
"x509",
|
||
"certificate signed by unknown authority",
|
||
"tls:",
|
||
)
|
||
_cert_manager_recovered = False
|
||
_last_error: str = ""
|
||
_apply_ok = False
|
||
for _attempt in range(1, 7): # up to 6 attempts; first triggers cert-manager recovery
|
||
r = _kubectl(["apply", "-f", url], env=env, timeout=120)
|
||
if r.returncode == 0:
|
||
if r.stdout.strip():
|
||
_log(log, r.stdout.strip())
|
||
_apply_ok = True
|
||
break
|
||
combined = ((r.stdout or "") + (r.stderr or "")).lower()
|
||
_last_error = (r.stderr or r.stdout or "").strip()
|
||
is_cert_issue = any(pat in combined for pat in _CERT_WEBHOOK_PATTERNS)
|
||
if is_cert_issue:
|
||
if not _cert_manager_recovered:
|
||
_log(log, "WARN: cert-manager webhook error; ensuring cert-manager and retrying...")
|
||
ca_ready = _ensure_cert_manager_for_barman(env=env, log=log)
|
||
_cert_manager_recovered = True
|
||
if ca_ready:
|
||
_log(log, "cert-manager recovered; webhook CA bundle ready.")
|
||
else:
|
||
_log(log, "WARN: cert-manager restarted but CA bundle not confirmed; proceeding with retries...")
|
||
else:
|
||
wait_s = min(20 * _attempt, 60)
|
||
_log(log, f"Webhook CA not yet trusted (attempt {_attempt}); retrying in {wait_s} s...")
|
||
time.sleep(wait_s)
|
||
else:
|
||
raise RuntimeError(
|
||
f"Failed to apply Barman Cloud plugin manifest: {_last_error}"
|
||
)
|
||
if not _apply_ok:
|
||
raise RuntimeError(
|
||
f"Failed to apply Barman Cloud plugin after cert-manager restart: {_last_error}"
|
||
)
|
||
|
||
# Wait for barman-cloud deployment if it was created
|
||
r3 = _kubectl(["-n", "cnpg-system", "get", "deploy", "barman-cloud"], env=env, timeout=15)
|
||
if r3.returncode == 0:
|
||
timeout_s = int((env or {}).get("BARMAN_DEPLOY_TIMEOUT", "300") or "300")
|
||
_kubectl(
|
||
["-n", "cnpg-system", "rollout", "status", "deploy/barman-cloud", f"--timeout={timeout_s}s"],
|
||
env=env, timeout=timeout_s + 10,
|
||
)
|
||
|
||
# Wait for cert-manager to issue the Barman TLS secrets that the pod mounts
|
||
tls_timeout = int((env or {}).get("BARMAN_TLS_TIMEOUT", "300") or "300")
|
||
if not _wait_for_barman_tls_secrets(env=env, log=log, timeout=tls_timeout):
|
||
_log(log, f"WARN: Barman Cloud TLS secrets not ready after {tls_timeout}s; bootstrapping self-signed fallback...")
|
||
_bootstrap_barman_tls_secrets(env=env, log=log)
|
||
|
||
|
||
def _ensure_cert_manager_for_barman(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> bool:
|
||
def _exists(kind: str, name: str) -> bool:
|
||
return _kubectl(["-n", "cert-manager", "get", kind, name], env=env, timeout=20).returncode == 0
|
||
|
||
need_bootstrap = not all(
|
||
[
|
||
_exists("service", "cert-manager-webhook"),
|
||
_exists("deployment", "cert-manager"),
|
||
_exists("deployment", "cert-manager-webhook"),
|
||
_exists("deployment", "cert-manager-cainjector"),
|
||
]
|
||
)
|
||
|
||
if need_bootstrap:
|
||
_log(log, "[CNPG] cert-manager resources missing; bootstrapping cert-manager...")
|
||
root = Path(
|
||
(env or {}).get("KNOE_HOME")
|
||
or (env or {}).get("KNOE_SERVICE")
|
||
or Path(__file__).resolve().parents[4]
|
||
)
|
||
init_script = root / "etc" / "init_certmgr.sh"
|
||
if init_script.exists():
|
||
cmd = ["bash", str(init_script)]
|
||
mode = str((env or {}).get("KNOE_MODE") or "").strip()
|
||
if mode:
|
||
cmd.extend(["--mode", mode])
|
||
cfg_hint = (
|
||
str((env or {}).get("KNOE_CONFIG_PATH") or "").strip()
|
||
or str((env or {}).get("PROLE_CFG_PATH") or "").strip()
|
||
or str((env or {}).get("PROLE_CFG") or "").strip()
|
||
)
|
||
if cfg_hint:
|
||
cmd.extend(["-c", cfg_hint])
|
||
cmd.append("start")
|
||
r = subprocess.run(
|
||
cmd,
|
||
env=env,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=600,
|
||
)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(
|
||
"Failed to bootstrap cert-manager via init_certmgr.sh: "
|
||
f"{(r.stderr or r.stdout or '').strip()}"
|
||
)
|
||
else:
|
||
_kubectl(["get", "namespace", "cert-manager"], env=env, timeout=20)
|
||
_kubectl(["create", "namespace", "cert-manager"], env=env, timeout=20)
|
||
certmgr_version = str((env or {}).get("CERTMGR_VERSION") or "v1.14.6").strip()
|
||
certmgr_url = (
|
||
"https://github.com/cert-manager/cert-manager/releases/download/"
|
||
f"{certmgr_version}/cert-manager.yaml"
|
||
)
|
||
r = _kubectl(["apply", "-f", certmgr_url], env=env, timeout=180)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(
|
||
f"Failed to bootstrap cert-manager from {certmgr_url}: "
|
||
f"{(r.stderr or r.stdout or '').strip()}"
|
||
)
|
||
|
||
for deploy in ["cert-manager", "cert-manager-webhook", "cert-manager-cainjector"]:
|
||
if _exists("deployment", deploy):
|
||
_kubectl(["-n", "cert-manager", "rollout", "restart", f"deploy/{deploy}"], env=env, timeout=45)
|
||
for deploy in ["cert-manager-webhook", "cert-manager", "cert-manager-cainjector"]:
|
||
if _exists("deployment", deploy):
|
||
_kubectl(
|
||
["-n", "cert-manager", "rollout", "status", f"deploy/{deploy}", "--timeout=240s"],
|
||
env=env,
|
||
timeout=250,
|
||
)
|
||
|
||
_log(log, "Waiting for cert-manager webhook CA bundle to be injected...")
|
||
if _wait_for_webhook_ca_bundle(env=env, log=log, timeout=300):
|
||
return True
|
||
_log(log, "WARN: cert-manager webhook CA bundle not detected within 300 s; attempting manual CA injection...")
|
||
if _try_inject_ca_bundle_manually(env=env, log=log):
|
||
_log(log, "Manual CA bundle injection succeeded.")
|
||
return True
|
||
_log(log, "WARN: Manual CA bundle injection also failed; cert-manager webhook may not be fully trusted.")
|
||
return False
|
||
|
||
|
||
def _try_inject_ca_bundle_manually(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> bool:
|
||
"""Manually patch the caBundle into the cert-manager ValidatingWebhookConfiguration.
|
||
|
||
Reads the CA certificate from the ``cert-manager-webhook-ca`` secret (created
|
||
by cert-manager itself) and patches it directly into the webhook configuration
|
||
when the cainjector is too slow or has not yet reconciled.
|
||
|
||
Returns True if the patch succeeded and the caBundle is now non-empty.
|
||
"""
|
||
for secret_name in ("cert-manager-webhook-ca", "cert-manager-cainjector-leader-election"):
|
||
r = _kubectl(
|
||
["-n", "cert-manager", "get", "secret", secret_name, "-o", "jsonpath={.data.tls\\.crt}"],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
ca_bundle = (r.stdout or "").strip()
|
||
if r.returncode == 0 and ca_bundle:
|
||
break
|
||
else:
|
||
_log(log, "WARN: cert-manager CA secret not found; skipping manual caBundle patch.")
|
||
return False
|
||
|
||
_log(log, "Patching cert-manager-webhook ValidatingWebhookConfiguration with CA bundle...")
|
||
patch_json = json.dumps(
|
||
[{"op": "replace", "path": "/webhooks/0/clientConfig/caBundle", "value": ca_bundle}]
|
||
)
|
||
r2 = _kubectl(
|
||
[
|
||
"patch",
|
||
"validatingwebhookconfiguration",
|
||
"cert-manager-webhook",
|
||
"--type=json",
|
||
f"-p={patch_json}",
|
||
],
|
||
env=env,
|
||
timeout=20,
|
||
)
|
||
if r2.returncode != 0:
|
||
_log(log, f"WARN: Failed to patch caBundle: {(r2.stderr or r2.stdout or '').strip()}")
|
||
return False
|
||
# Brief wait for API server to propagate the patch
|
||
time.sleep(3)
|
||
return _wait_for_webhook_ca_bundle(env=env, log=log, timeout=30)
|
||
|
||
|
||
def _wait_for_webhook_ca_bundle(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
timeout: int = 300,
|
||
) -> bool:
|
||
"""Poll until the cert-manager ValidatingWebhookConfiguration has a CA bundle injected.
|
||
|
||
Returns True if the caBundle field becomes non-empty within *timeout* seconds,
|
||
False otherwise.
|
||
"""
|
||
deadline = time.time() + timeout
|
||
interval = 5
|
||
while time.time() < deadline:
|
||
r = _kubectl(
|
||
[
|
||
"get",
|
||
"validatingwebhookconfiguration",
|
||
"cert-manager-webhook",
|
||
"-o",
|
||
"jsonpath={.webhooks[0].clientConfig.caBundle}",
|
||
],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
return True
|
||
time.sleep(interval)
|
||
return False
|
||
|
||
|
||
def _wait_for_barman_tls_secrets(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
timeout: int = 180,
|
||
) -> bool:
|
||
"""Poll until cert-manager has issued both Barman Cloud TLS secrets in cnpg-system.
|
||
|
||
The Barman Cloud plugin manifest declares two Certificate resources
|
||
(``barman-cloud-client`` and ``barman-cloud-server``). cert-manager
|
||
populates the corresponding secrets asynchronously; the pod cannot start
|
||
until both secrets exist and have non-empty ``tls.crt`` / ``tls.key`` data.
|
||
|
||
Returns True when all four fields are present, False on timeout.
|
||
"""
|
||
_log(log, f"Waiting for Barman Cloud TLS secrets in cnpg-system (timeout: {timeout}s)...")
|
||
deadline = time.time() + timeout
|
||
interval = 5
|
||
while time.time() < deadline:
|
||
all_ready = True
|
||
for secret in ("barman-cloud-client-tls", "barman-cloud-server-tls"):
|
||
for field in ("tls.crt", "tls.key"):
|
||
escaped_field = field.replace(".", "\\.")
|
||
r = _kubectl(
|
||
[
|
||
"-n",
|
||
"cnpg-system",
|
||
"get",
|
||
"secret",
|
||
secret,
|
||
"-o",
|
||
f"jsonpath={{.data.{escaped_field}}}",
|
||
],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
if r.returncode != 0 or not (r.stdout or "").strip():
|
||
all_ready = False
|
||
break
|
||
if not all_ready:
|
||
break
|
||
if all_ready:
|
||
_log(log, "Barman Cloud TLS secrets are available.")
|
||
return True
|
||
time.sleep(interval)
|
||
_log(log, f"WARN: Barman Cloud TLS secrets not ready after {timeout}s.")
|
||
return False
|
||
|
||
|
||
def _bootstrap_barman_tls_secrets(
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Generate and apply self-signed TLS secrets for the Barman Cloud plugin.
|
||
|
||
Fallback for environments where cert-manager cannot issue the Certificate
|
||
resources in time (e.g. GKE Autopilot with slow cert-manager scheduling).
|
||
Generates EC P-256 self-signed certs and applies them as
|
||
``kubernetes.io/tls`` secrets directly to ``cnpg-system``, unblocking
|
||
the barman-cloud pod's volume mounts.
|
||
|
||
Secrets managed (both ``kubernetes.io/tls`` type):
|
||
- ``barman-cloud-server-tls``
|
||
- ``barman-cloud-client-tls``
|
||
"""
|
||
namespace = "cnpg-system"
|
||
server_secret = "barman-cloud-server-tls"
|
||
client_secret = "barman-cloud-client-tls"
|
||
|
||
def _secret_has_tls(name: str) -> bool:
|
||
for field in ("tls.crt", "tls.key"):
|
||
escaped_field = field.replace(".", "\\.")
|
||
r = _kubectl(
|
||
["-n", namespace, "get", "secret", name, "-o", f"jsonpath={{.data.{escaped_field}}}"],
|
||
env=env,
|
||
timeout=15,
|
||
)
|
||
if r.returncode != 0 or not (r.stdout or "").strip():
|
||
return False
|
||
return True
|
||
|
||
if _secret_has_tls(server_secret) and _secret_has_tls(client_secret):
|
||
_log(log, "Barman Cloud TLS secrets already present — skipping bootstrap.")
|
||
return
|
||
|
||
_log(log, "Bootstrapping Barman Cloud TLS secrets in cnpg-system (cert-manager fallback)...")
|
||
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
one_year = datetime.timedelta(days=365)
|
||
|
||
ca_key = generate_private_key(SECP256R1())
|
||
ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "barman-cloud CA")])
|
||
ca_cert = (
|
||
x509.CertificateBuilder()
|
||
.subject_name(ca_name)
|
||
.issuer_name(ca_name)
|
||
.public_key(ca_key.public_key())
|
||
.serial_number(x509.random_serial_number())
|
||
.not_valid_before(now)
|
||
.not_valid_after(now + one_year)
|
||
.add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False)
|
||
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
|
||
.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
|
||
.sign(ca_key, hashes.SHA256())
|
||
)
|
||
|
||
def _pem(obj) -> str:
|
||
return obj.public_bytes(serialization.Encoding.PEM).decode()
|
||
|
||
def _key_pem(k) -> str:
|
||
return k.private_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PrivateFormat.TraditionalOpenSSL,
|
||
serialization.NoEncryption(),
|
||
).decode()
|
||
|
||
def _make_cert(cn: str, sans: list[str]):
|
||
key = generate_private_key(SECP256R1())
|
||
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
|
||
cert = (
|
||
x509.CertificateBuilder()
|
||
.subject_name(name)
|
||
.issuer_name(ca_name)
|
||
.public_key(key.public_key())
|
||
.serial_number(x509.random_serial_number())
|
||
.not_valid_before(now)
|
||
.not_valid_after(now + one_year)
|
||
.add_extension(
|
||
x509.SubjectAlternativeName([x509.DNSName(d) for d in sans]),
|
||
critical=False,
|
||
)
|
||
.add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
|
||
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
|
||
.sign(ca_key, hashes.SHA256())
|
||
)
|
||
return cert, key
|
||
|
||
def _apply_tls_secret(name: str, cert_pem: str, key_pem: str) -> None:
|
||
manifest = (
|
||
"apiVersion: v1\n"
|
||
"kind: Secret\n"
|
||
"type: kubernetes.io/tls\n"
|
||
"metadata:\n"
|
||
f" name: {name}\n"
|
||
f" namespace: {namespace}\n"
|
||
"data:\n"
|
||
f" tls.crt: {base64.b64encode(cert_pem.encode()).decode()}\n"
|
||
f" tls.key: {base64.b64encode(key_pem.encode()).decode()}\n"
|
||
)
|
||
res = _kubectl_run(
|
||
["apply", "-n", namespace, "-f", "-"],
|
||
env=env,
|
||
input_text=manifest,
|
||
)
|
||
if res.returncode != 0:
|
||
raise RuntimeError(f"Failed to apply Barman TLS secret '{name}': {res.stderr}")
|
||
_log(log, f"Applied Barman TLS secret '{name}' in '{namespace}'.")
|
||
|
||
srv_cert, srv_key = _make_cert(
|
||
f"barman-cloud.{namespace}.svc",
|
||
[
|
||
"barman-cloud",
|
||
f"barman-cloud.{namespace}.svc",
|
||
f"barman-cloud.{namespace}.svc.cluster.local",
|
||
],
|
||
)
|
||
_apply_tls_secret(server_secret, _pem(srv_cert), _key_pem(srv_key))
|
||
|
||
cli_cert, cli_key = _make_cert(
|
||
"barman-cloud-client",
|
||
["barman-cloud-client"],
|
||
)
|
||
_apply_tls_secret(client_secret, _pem(cli_cert), _key_pem(cli_key))
|
||
|
||
|
||
def _resolve_barman_plugin_tag(env: dict | None, log: _LogFn | None) -> str:
|
||
"""Fetch the latest Barman Cloud plugin release tag from GitHub, with fallback."""
|
||
fallback = ((env or {}).get("BARMAN_PLUGIN_FALLBACK_VERSION") or _BARMAN_PLUGIN_FALLBACK_VERSION).strip()
|
||
try:
|
||
import urllib.request as _req
|
||
import json as _json
|
||
with _req.urlopen(
|
||
"https://api.github.com/repos/cloudnative-pg/plugin-barman-cloud/releases/latest",
|
||
timeout=10,
|
||
) as resp:
|
||
data = _json.loads(resp.read())
|
||
tag = (data.get("tag_name") or "").strip()
|
||
if tag and tag != "null":
|
||
return tag
|
||
except Exception as exc:
|
||
_log(log, f"WARN: could not fetch latest Barman plugin version: {exc}; using v{fallback}")
|
||
return f"v{fallback}"
|
||
|
||
|
||
def _bootstrap_db_user_secrets(
|
||
namespace: str,
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Create knoe-db-user and knoe-db-superuser secrets from DB_PASSWORD if missing.
|
||
|
||
Uses ``env["DB_PASSWORD"]`` and ``env.get("KNOE_DB_USER", "postgres")`` as the
|
||
credentials source. Skips silently if the secrets already exist or if no
|
||
password is available in env.
|
||
"""
|
||
pw = ((env or {}).get("DB_PASSWORD") or "").strip()
|
||
if not pw or pw.startswith("${"):
|
||
_log(log, "WARN: DB_PASSWORD not resolved in env; skipping db-user secret bootstrap.")
|
||
return
|
||
|
||
# CNPG requires the secret username to match the cluster owner field (owner: knoe).
|
||
# KNOE_DB_USER is the admin/superuser login name, not the cluster owner.
|
||
db_owner = ((env or {}).get("CNPG_DB_OWNER") or "knoe").strip()
|
||
|
||
def _apply_if_missing(secret_name: str, username: str, password: str) -> None:
|
||
if _kubectl_ok(["-n", namespace, "get", "secret", secret_name], env=env):
|
||
return
|
||
_log(log, f"Creating missing secret '{secret_name}' in '{namespace}'...")
|
||
manifest_res = _kubectl(
|
||
[
|
||
"create",
|
||
"secret",
|
||
"generic",
|
||
secret_name,
|
||
"-n",
|
||
namespace,
|
||
f"--from-literal=username={username}",
|
||
f"--from-literal=password={password}",
|
||
"--dry-run=client",
|
||
"-o",
|
||
"yaml",
|
||
],
|
||
env=env,
|
||
)
|
||
if manifest_res.returncode != 0:
|
||
raise RuntimeError(f"Failed to render secret '{secret_name}':\n{manifest_res.stderr}")
|
||
apply_res = _kubectl_run(
|
||
["apply", "-n", namespace, "-f", "-"],
|
||
env=env,
|
||
input_text=manifest_res.stdout,
|
||
)
|
||
if apply_res.returncode != 0:
|
||
raise RuntimeError(f"Failed to apply secret '{secret_name}':\n{apply_res.stderr}")
|
||
_log(log, f"Created secret '{secret_name}' in '{namespace}'.")
|
||
|
||
_apply_if_missing("knoe-db-user", db_owner, pw)
|
||
_apply_if_missing("knoe-db-superuser", "postgres", pw)
|
||
|
||
|
||
def _bootstrap_cnpg_admin_key(
|
||
namespace: str,
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Generate and apply cnpg-admin-key secret (admin.key + admin.pub) if missing.
|
||
|
||
Uses ED25519 if available; falls back to RSA-4096. The secret is
|
||
create-if-missing: existing keys are never rotated.
|
||
"""
|
||
secret_name = "cnpg-admin-key"
|
||
if _kubectl_ok(["-n", namespace, "get", "secret", secret_name], env=env):
|
||
return
|
||
|
||
_log(log, f"Creating missing secret '{secret_name}' in '{namespace}'...")
|
||
|
||
# Generate ED25519 keypair; fall back to RSA-4096 if the runtime lacks it.
|
||
try:
|
||
priv_key = Ed25519PrivateKey.generate()
|
||
priv_pem = priv_key.private_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PrivateFormat.PKCS8,
|
||
serialization.NoEncryption(),
|
||
).decode()
|
||
pub_pem = priv_key.public_key().public_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||
).decode()
|
||
except Exception: # pragma: no cover — runtime without ED25519 support
|
||
from cryptography.hazmat.primitives.asymmetric import rsa as _rsa
|
||
priv_key_rsa = _rsa.generate_private_key(public_exponent=65537, key_size=4096)
|
||
priv_pem = priv_key_rsa.private_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PrivateFormat.PKCS8,
|
||
serialization.NoEncryption(),
|
||
).decode()
|
||
pub_pem = priv_key_rsa.public_key().public_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||
).decode()
|
||
|
||
manifest_res = _kubectl(
|
||
[
|
||
"create",
|
||
"secret",
|
||
"generic",
|
||
secret_name,
|
||
"-n",
|
||
namespace,
|
||
f"--from-literal=admin.key={priv_pem}",
|
||
f"--from-literal=admin.pub={pub_pem}",
|
||
"--dry-run=client",
|
||
"-o",
|
||
"yaml",
|
||
],
|
||
env=env,
|
||
)
|
||
if manifest_res.returncode != 0:
|
||
raise RuntimeError(f"Failed to render secret '{secret_name}':\n{manifest_res.stderr}")
|
||
apply_res = _kubectl_run(
|
||
["apply", "-n", namespace, "-f", "-"],
|
||
env=env,
|
||
input_text=manifest_res.stdout,
|
||
)
|
||
if apply_res.returncode != 0:
|
||
raise RuntimeError(f"Failed to apply secret '{secret_name}':\n{apply_res.stderr}")
|
||
_log(log, f"Created secret '{secret_name}' in '{namespace}'.")
|
||
|
||
|
||
def bootstrap_cnpg_tls_secrets(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Generate and apply self-signed CA + TLS secrets for CNPG if missing.
|
||
|
||
Creates two Kubernetes secrets (skipped if already present):
|
||
- ``{cluster_name}-ca`` : Opaque, keys ``ca.crt`` and ``ca.key``
|
||
- ``{cluster_name}-tls`` : kubernetes.io/tls, keys ``tls.crt`` and ``tls.key``
|
||
|
||
The CA uses CN=``{cluster_name} CNPG CA`` and is self-signed (EC P-256, SHA-256).
|
||
The server cert is signed by that CA and carries SANs matching the standard
|
||
CNPG service names for the cluster.
|
||
"""
|
||
ca_secret = f"{cluster_name}-ca"
|
||
tls_secret = f"{cluster_name}-tls"
|
||
|
||
# Ensure namespace exists before any secret operations
|
||
_ensure_namespace(namespace, env=env)
|
||
|
||
ca_exists = _kubectl_ok(["-n", namespace, "get", "secret", ca_secret], env=env)
|
||
tls_exists = _kubectl_ok(["-n", namespace, "get", "secret", tls_secret], env=env)
|
||
if ca_exists and tls_exists:
|
||
_log(log, f"CNPG TLS secrets already present in '{namespace}' — skipping bootstrap.")
|
||
return
|
||
|
||
_log(log, f"Bootstrapping CNPG TLS secrets in namespace '{namespace}'...")
|
||
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
one_year = datetime.timedelta(days=365)
|
||
|
||
# --- CA ---
|
||
ca_key = generate_private_key(SECP256R1())
|
||
ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"{cluster_name} CNPG CA")])
|
||
ca_cert = (
|
||
x509.CertificateBuilder()
|
||
.subject_name(ca_name)
|
||
.issuer_name(ca_name)
|
||
.public_key(ca_key.public_key())
|
||
.serial_number(x509.random_serial_number())
|
||
.not_valid_before(now)
|
||
.not_valid_after(now + one_year)
|
||
.add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False)
|
||
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
|
||
.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
|
||
.sign(ca_key, hashes.SHA256())
|
||
)
|
||
|
||
# --- Server cert ---
|
||
srv_key = generate_private_key(SECP256R1())
|
||
san_dns = [
|
||
cluster_name,
|
||
f"{cluster_name}-rw",
|
||
f"{cluster_name}-rw.{namespace}.svc",
|
||
f"{cluster_name}-r",
|
||
f"{cluster_name}-ro",
|
||
]
|
||
primary_srv_cn = f"{cluster_name}.{namespace}.svc"
|
||
srv_cn = primary_srv_cn
|
||
if len(srv_cn) > 64:
|
||
for candidate in (f"{cluster_name}-rw", cluster_name, "cnpg-server"):
|
||
if len(candidate) <= 64:
|
||
srv_cn = candidate
|
||
break
|
||
_log(
|
||
log,
|
||
"CNPG TLS subject CN exceeds X.509 limit; "
|
||
f"using fallback CN '{srv_cn}' instead of '{primary_srv_cn}'.",
|
||
)
|
||
|
||
srv_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, srv_cn)])
|
||
srv_cert = (
|
||
x509.CertificateBuilder()
|
||
.subject_name(srv_name)
|
||
.issuer_name(ca_name)
|
||
.public_key(srv_key.public_key())
|
||
.serial_number(x509.random_serial_number())
|
||
.not_valid_before(now)
|
||
.not_valid_after(now + one_year)
|
||
.add_extension(
|
||
x509.SubjectAlternativeName([x509.DNSName(d) for d in san_dns]),
|
||
critical=False,
|
||
)
|
||
.add_extension(x509.SubjectKeyIdentifier.from_public_key(srv_key.public_key()), critical=False)
|
||
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
|
||
.sign(ca_key, hashes.SHA256())
|
||
)
|
||
|
||
def _pem(obj) -> str:
|
||
return obj.public_bytes(serialization.Encoding.PEM).decode()
|
||
|
||
def _key_pem(k) -> str:
|
||
return k.private_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PrivateFormat.TraditionalOpenSSL,
|
||
serialization.NoEncryption(),
|
||
).decode()
|
||
|
||
def _apply_secret(manifest: str) -> None:
|
||
res = _kubectl_run(
|
||
["apply", "-n", namespace, "-f", "-"],
|
||
env=env,
|
||
input_text=manifest,
|
||
)
|
||
if res.returncode != 0:
|
||
raise RuntimeError(f"kubectl apply failed:\n{res.stderr}")
|
||
|
||
ca_crt_pem = _pem(ca_cert)
|
||
ca_key_pem = _key_pem(ca_key)
|
||
|
||
# Build CA secret manifest (Opaque)
|
||
ca_manifest = (
|
||
"apiVersion: v1\n"
|
||
"kind: Secret\n"
|
||
"metadata:\n"
|
||
f" name: {ca_secret}\n"
|
||
f" namespace: {namespace}\n"
|
||
"data:\n"
|
||
f" ca.crt: {base64.b64encode(ca_crt_pem.encode()).decode()}\n"
|
||
f" ca.key: {base64.b64encode(ca_key_pem.encode()).decode()}\n"
|
||
)
|
||
_apply_secret(ca_manifest)
|
||
_log(log, f"Applied secret '{ca_secret}' in '{namespace}'.")
|
||
|
||
tls_crt_pem = _pem(srv_cert)
|
||
tls_key_pem = _key_pem(srv_key)
|
||
|
||
# Build TLS secret manifest (kubernetes.io/tls)
|
||
tls_manifest = (
|
||
"apiVersion: v1\n"
|
||
"kind: Secret\n"
|
||
"type: kubernetes.io/tls\n"
|
||
"metadata:\n"
|
||
f" name: {tls_secret}\n"
|
||
f" namespace: {namespace}\n"
|
||
"data:\n"
|
||
f" tls.crt: {base64.b64encode(tls_crt_pem.encode()).decode()}\n"
|
||
f" tls.key: {base64.b64encode(tls_key_pem.encode()).decode()}\n"
|
||
)
|
||
_apply_secret(tls_manifest)
|
||
_log(log, f"Applied secret '{tls_secret}' in '{namespace}'.")
|
||
|
||
|
||
def initialize(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None = None,
|
||
project_root: str | Path = ".",
|
||
log: _LogFn | None = None,
|
||
mode: str = "",
|
||
) -> None:
|
||
"""
|
||
Full CNPG initialization pipeline.
|
||
|
||
Steps (matching init_cloudnative_pg.sh initialize):
|
||
1. Ensure the CNPG operator is installed and ready
|
||
2. Pin the controller to storage nodes (k3s)
|
||
3. Install the Barman Cloud plugin
|
||
4. Preflight: verify required secrets exist
|
||
5. Apply the CNPG cluster manifest
|
||
6. Wait for cluster pods to become Ready
|
||
"""
|
||
_log(log, f"Initializing CNPG cluster '{cluster_name}' in namespace '{namespace}'...")
|
||
|
||
# Fast-path: if cluster already healthy and all pods Ready, skip the full pipeline.
|
||
# install_barman_plugin() fetches from GitHub (up to 6 × 120 s attempts) — pure waste
|
||
# when the cluster is already running. We check phase then pod readiness.
|
||
_phase_r = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.status.phase}"],
|
||
env=env, timeout=15,
|
||
)
|
||
if _phase_r.returncode == 0 and "healthy" in (_phase_r.stdout or "").lower():
|
||
_pods_r = _kubectl(
|
||
["-n", namespace, "get", "pods", "--no-headers",
|
||
"-l", f"cnpg.io/cluster={cluster_name}"],
|
||
env=env, timeout=20,
|
||
)
|
||
_pod_lines = [ln for ln in (_pods_r.stdout or "").splitlines() if ln.strip()]
|
||
_not_ready = [
|
||
ln for ln in _pod_lines
|
||
if len(ln.split()) < 2 or not ln.split()[1].startswith("2/")
|
||
]
|
||
if _pod_lines and not _not_ready:
|
||
_log(log, f"[CNPG] Cluster '{cluster_name}' already healthy with all pods Ready — skipping reinitialize")
|
||
return
|
||
|
||
_log(log, "[CNPG] Ensuring operator is installed and healthy...")
|
||
ensure_operator(
|
||
env=env,
|
||
log=log,
|
||
mode=mode,
|
||
allow_reapply_if_unhealthy=True,
|
||
aggressive_cleanup=True,
|
||
)
|
||
_log(log, "[CNPG] Pinning controller placement...")
|
||
pin_controller(env=env, log=log, mode=mode)
|
||
_log(log, "[CNPG] Installing Barman Cloud plugin...")
|
||
install_barman_plugin(env=env, log=log)
|
||
|
||
# Bootstrap TLS secrets (create-if-missing; no rotation of existing certs)
|
||
_log(log, "[CNPG] Bootstrapping TLS secrets...")
|
||
bootstrap_cnpg_tls_secrets(namespace=namespace, cluster_name=cluster_name, env=env, log=log)
|
||
|
||
# Bootstrap DB user secrets from env if missing (create-if-missing only)
|
||
_log(log, "[CNPG] Ensuring DB user secrets are present...")
|
||
_bootstrap_db_user_secrets(namespace=namespace, env=env, log=log)
|
||
|
||
# Bootstrap admin keypair secret (create-if-missing; no rotation)
|
||
_log(log, "[CNPG] Ensuring admin keypair secret is present...")
|
||
_bootstrap_cnpg_admin_key(namespace=namespace, env=env, log=log)
|
||
|
||
# Preflight: required secrets must exist before applying the cluster
|
||
missing = []
|
||
for secret in ["knoe-db-user", f"{cluster_name}-tls", f"{cluster_name}-ca"]:
|
||
if not _kubectl_ok(["-n", namespace, "get", "secret", secret], env=env):
|
||
missing.append(secret)
|
||
if missing:
|
||
raise RuntimeError(
|
||
f"Required secrets missing in namespace '{namespace}': {', '.join(missing)}. "
|
||
"The CNPG cluster cannot start without these secrets."
|
||
)
|
||
_log(log, "Pre-flight check passed: all required secrets present.")
|
||
# Release any Released PVs so new PVCs can bind (idempotent, k3s only)
|
||
if mode == "k3s":
|
||
_reconcile_unbound_synology_data_wal_pvs(env=env, log=log)
|
||
_release_stale_pv_claims(env=env, log=log)
|
||
_remove_legacy_barman_object_store_when_plugin_is_enabled(
|
||
namespace=namespace,
|
||
cluster_name=cluster_name,
|
||
env=env,
|
||
log=log,
|
||
)
|
||
|
||
# Recovery: Detect and clean up stuck cluster bootstrap
|
||
_check_and_fix_broken_bootstrap(namespace, cluster_name, env, log)
|
||
|
||
_log(log, "[CNPG] Applying cluster manifest...")
|
||
manifest = _resolve_cnpg_manifest(project_root, env)
|
||
_apply_manifest(namespace, manifest, env, log)
|
||
|
||
wait_timeout = int((env or {}).get("CNPG_WAIT_TIMEOUT", "900") or "900")
|
||
_log(log, f"[CNPG] Waiting for cluster pods to become Ready (timeout={wait_timeout}s)...")
|
||
_wait_cnpg_pods(namespace, cluster_name, env, timeout=wait_timeout, log=log)
|
||
|
||
_log(log, f"Initialization complete for CNPG cluster '{cluster_name}'.")
|
||
|
||
|
||
def _check_and_fix_broken_bootstrap(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""Detect and clean up a CNPG cluster stuck in a broken bootstrap state.
|
||
|
||
This covers cases like ImagePullBackOff or ErrImagePull on the first node,
|
||
which can prevent the cluster from ever reaching a stable state even after
|
||
the image is fixed, unless the initial failed attempt is cleared.
|
||
"""
|
||
# 1. Check if cluster exists
|
||
r = _kubectl(["-n", namespace, "get", "cluster", cluster_name, "-o", "json"], env=env, timeout=10)
|
||
if r.returncode != 0:
|
||
return # Cluster doesn't exist yet, clean slate.
|
||
|
||
try:
|
||
cluster_json = json.loads(r.stdout or "{}")
|
||
except Exception:
|
||
return
|
||
|
||
# 2. Check cluster status
|
||
status = cluster_json.get("status") or {}
|
||
phase = status.get("phase", "")
|
||
instances = status.get("instances", 0)
|
||
ready_instances = status.get("readyInstances", 0)
|
||
|
||
# If cluster is already healthy and has instances, don't touch it.
|
||
if phase == "Healthy" and ready_instances > 0:
|
||
return
|
||
|
||
# 3. Look for stuck pods
|
||
r_pods = _kubectl(
|
||
["-n", namespace, "get", "pods", "-l", f"cnpg.io/cluster={cluster_name}", "-o", "json"],
|
||
env=env, timeout=15
|
||
)
|
||
if r_pods.returncode != 0:
|
||
return
|
||
|
||
try:
|
||
pods_json = json.loads(r_pods.stdout or "{}")
|
||
except Exception:
|
||
return
|
||
|
||
items = pods_json.get("items", [])
|
||
if not items:
|
||
# No pods yet, maybe it's just slow or just created.
|
||
return
|
||
|
||
stuck_reasons = {"ImagePullBackOff", "ErrImagePull", "InvalidImageName", "CreateContainerConfigError"}
|
||
is_stuck = False
|
||
reasons = set()
|
||
|
||
for pod in items:
|
||
pod_status = pod.get("status") or {}
|
||
container_statuses = pod_status.get("containerStatuses") or []
|
||
for cs in container_statuses:
|
||
waiting = cs.get("state", {}).get("waiting") or {}
|
||
reason = waiting.get("reason", "")
|
||
if reason in stuck_reasons:
|
||
is_stuck = True
|
||
reasons.add(reason)
|
||
|
||
if is_stuck:
|
||
_log(log, f"[RECOVERY] CNPG cluster '{cluster_name}' is stuck in {', '.join(reasons)}. Deleting for clean retry.")
|
||
# Delete the cluster. We use --wait=false to avoid hanging if the operator is busy,
|
||
# though standard delete is usually fine.
|
||
_kubectl(["-n", namespace, "delete", "cluster", cluster_name, "--wait=true", "--timeout=60s"], env=env, timeout=70)
|
||
_log(log, f"[RECOVERY] Cluster '{cluster_name}' deleted. It will be recreated by the deploy flow.")
|
||
|
||
|
||
def deploy(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None = None,
|
||
project_root: str | Path = ".",
|
||
log: _LogFn | None = None,
|
||
mode: str = "",
|
||
) -> None:
|
||
"""
|
||
Image-aware CNPG deploy pipeline.
|
||
|
||
Steps (matching init_cloudnative_pg.sh deploy_cluster):
|
||
1. Ensure the CNPG operator is installed and ready
|
||
2. Pin the controller to storage nodes (k3s)
|
||
3. Install the Barman Cloud plugin
|
||
4. Apply the CNPG cluster manifest
|
||
5. Reconcile instance count to match eligible node count
|
||
"""
|
||
_log(log, f"Deploying CNPG cluster '{cluster_name}' in namespace '{namespace}'...")
|
||
|
||
# Ensure target namespace exists before applying/updating Cluster resources.
|
||
_ensure_namespace(namespace, env=env)
|
||
|
||
ensure_operator(
|
||
env=env,
|
||
log=log,
|
||
mode=mode,
|
||
allow_reapply_if_unhealthy=False,
|
||
aggressive_cleanup=False,
|
||
)
|
||
pin_controller(env=env, log=log, mode=mode)
|
||
install_barman_plugin(env=env, log=log)
|
||
_remove_legacy_barman_object_store_when_plugin_is_enabled(
|
||
namespace=namespace,
|
||
cluster_name=cluster_name,
|
||
env=env,
|
||
log=log,
|
||
)
|
||
|
||
# Recovery: Check for broken bootstrap (e.g. ImagePullBackOff on first init)
|
||
# and delete the cluster to allow a clean retry if requested/needed.
|
||
_check_and_fix_broken_bootstrap(namespace, cluster_name, env, log)
|
||
|
||
manifest = _resolve_cnpg_manifest(project_root, env)
|
||
_apply_manifest(namespace, manifest, env, log)
|
||
_reconcile_instances(namespace, cluster_name, env, log)
|
||
|
||
_log(log, f"Deploy complete for CNPG cluster '{cluster_name}'.")
|
||
|
||
|
||
def rollout(
|
||
namespace: str,
|
||
cluster_name: str,
|
||
env: dict | None = None,
|
||
log: _LogFn | None = None,
|
||
) -> None:
|
||
"""
|
||
Rolling restart of a CNPG cluster via ordered pod deletion.
|
||
|
||
Steps (matching init_cloudnative_pg.sh rollout_cluster):
|
||
1. Identify the current primary pod
|
||
2. Delete replica pods one-by-one, waiting for readiness
|
||
3. Promote a replica to primary (if kubectl-cnpg plugin is available)
|
||
4. Delete and recreate the old primary pod
|
||
"""
|
||
_log(log, f"Starting rollout for CNPG cluster '{cluster_name}' in namespace '{namespace}'...")
|
||
|
||
# Identify current primary
|
||
r = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.status.currentPrimary}"],
|
||
env=env, timeout=15,
|
||
)
|
||
primary = (r.stdout or "").strip()
|
||
if not primary:
|
||
raise RuntimeError(f"Could not identify primary instance for cluster '{cluster_name}'")
|
||
_log(log, f"Primary instance: {primary}")
|
||
|
||
# List all cluster pods
|
||
r2 = _kubectl(
|
||
["-n", namespace, "get", "pods", "-l", f"cnpg.io/cluster={cluster_name}",
|
||
"-o", "jsonpath={.items[*].metadata.name}"],
|
||
env=env, timeout=15,
|
||
)
|
||
pods = (r2.stdout or "").strip().split()
|
||
if not pods:
|
||
raise RuntimeError(f"No pods found for cluster '{cluster_name}'")
|
||
|
||
has_plugin = _kubectl_ok(["cnpg", "version"], env=env, timeout=10)
|
||
|
||
# Delete non-primary pods first
|
||
new_primary: str | None = None
|
||
for pod in pods:
|
||
if pod == primary:
|
||
continue
|
||
_log(log, f"Recreating non-primary pod: {pod}...")
|
||
_kubectl(["-n", namespace, "delete", "pod", pod], env=env, timeout=60)
|
||
_wait_pod_ready(namespace, pod, env, timeout=300, log=log)
|
||
if new_primary is None:
|
||
new_primary = pod
|
||
|
||
# Promote a replica if the cnpg plugin is available
|
||
if new_primary and has_plugin:
|
||
_log(log, f"Promoting {new_primary}...")
|
||
_kubectl(["cnpg", "promote", cluster_name, new_primary, "-n", namespace], env=env, timeout=30)
|
||
deadline = time.monotonic() + 300
|
||
while time.monotonic() < deadline:
|
||
r3 = _kubectl(
|
||
["-n", namespace, "get", "cluster", cluster_name,
|
||
"-o", "jsonpath={.status.currentPrimary}"],
|
||
env=env, timeout=10,
|
||
)
|
||
if (r3.stdout or "").strip() == new_primary:
|
||
_log(log, f"{new_primary} is now the primary.")
|
||
break
|
||
time.sleep(5)
|
||
elif new_primary and not has_plugin:
|
||
_log(log, "WARN: kubectl cnpg plugin not available; skipping explicit promotion.")
|
||
|
||
# Delete the old primary last
|
||
_log(log, f"Recreating old primary pod: {primary}...")
|
||
_kubectl(["-n", namespace, "delete", "pod", "-n", namespace, primary], env=env, timeout=60)
|
||
_wait_pod_ready(namespace, primary, env, timeout=300, log=log)
|
||
|
||
if has_plugin:
|
||
_log(log, "Rollout complete. Final cluster status:")
|
||
_kubectl(["cnpg", "status", cluster_name, "-n", namespace], env=env, timeout=30)
|
||
else:
|
||
_log(log, "Rollout complete.")
|