mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
c94c62e1bb
- persist and load DB master password via Ansible Vault bootstrap flow - enforce knoe-system service namespace and explicit app/db kubecontext targeting - improve OpenBao/CNPG deploy reliability and logging; add retries/readiness diagnostics - tighten reset/delete cluster behavior and expand installer/deploy pipeline test coverage Co-authored-by: Junie <junie@jetbrains.com>
1861 lines
68 KiB
Python
1861 lines
68 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("PROLE_HOME")
|
|
or (env or {}).get("PROLE_SERVICE")
|
|
or Path(__file__).resolve().parents[4]
|
|
)
|
|
manifest = root / "k8s" / "prole" / "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("PROLE_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" / "prole" / "knoe-db.yaml",
|
|
root / "k8s" / "prole" / "knoe-db.yaml",
|
|
]
|
|
if env:
|
|
prole_home = (env.get("PROLE_HOME") or "").strip()
|
|
if prole_home:
|
|
ph = Path(prole_home)
|
|
if mode == "k8s":
|
|
candidates.insert(0, ph / "deploy" / "gcp" / "gke" / "knoe-db.yaml")
|
|
candidates.append(ph / "k8s" / "prole" / "knoe-db.yaml")
|
|
for candidate in candidates:
|
|
if candidate.exists():
|
|
return candidate
|
|
return root / "k8s" / "prole" / "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("PROLE_MODE") or "").strip() or "k8s"
|
|
|
|
candidate_roots: list[Path] = []
|
|
prole_home = str((env or {}).get("PROLE_HOME") or "").strip()
|
|
if prole_home:
|
|
candidate_roots.append(Path(prole_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 _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)
|
|
|
|
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("PROLE_HOME")
|
|
or (env or {}).get("PROLE_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("PROLE_MODE") or "").strip()
|
|
if mode:
|
|
cmd.extend(["--mode", mode])
|
|
cfg_hint = (
|
|
str((env or {}).get("PROLE_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"):
|
|
r = _kubectl(
|
|
[
|
|
"-n",
|
|
"cnpg-system",
|
|
"get",
|
|
"secret",
|
|
secret,
|
|
"-o",
|
|
f"jsonpath={{.data.{field.replace('.', '\\.')}}}",
|
|
],
|
|
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"):
|
|
r = _kubectl(
|
|
["-n", namespace, "get", "secret", name, "-o", f"jsonpath={{.data.{field.replace('.', '\\.')}}}"],
|
|
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: prole).
|
|
# KNOE_DB_USER is the admin/superuser login name, not the cluster owner.
|
|
db_owner = ((env or {}).get("CNPG_DB_OWNER") or "prole").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}'...")
|
|
|
|
_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,
|
|
)
|
|
_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 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_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,
|
|
)
|
|
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.")
|