mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
be6ff2f1a8
- Added `supabase_collect_stale_retained_pv_info` for clearer separation of logic and improved reusability. - Enhanced zone detection for retained PVs using regex-based fallback. - Updated cleanup logic to auto-enable `SUPABASE_AUTO_CLEAN_RETAINED_PVS` and streamline handling of GCE disk deletion. - Introduced verification and retry mechanism to ensure stale PV artifacts are fully removed. - Improved error handling for remaining GCE disk cleanup with detailed remediation instructions.
462 lines
20 KiB
Bash
Executable File
462 lines
20 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
|
|
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
|
CONFIG_PATH="${PROLE_DEPLOY_CFG:-${ROOT_DIR}/conf/gke.cfg}"
|
|
|
|
if [[ -x "${ROOT_DIR}/.venv/bin/python3" ]]; then
|
|
PYTHON_BIN="${ROOT_DIR}/.venv/bin/python3"
|
|
elif [[ -x "${ROOT_DIR}/bin/python3" ]]; then
|
|
PYTHON_BIN="${ROOT_DIR}/bin/python3"
|
|
else
|
|
PYTHON_BIN="${PYTHON:-python3}"
|
|
fi
|
|
|
|
# Validate StorageClass on GKE before deployment
|
|
if [[ "$CONFIG_PATH" == *"gke.cfg"* ]]; then
|
|
echo "Validating Supabase StorageClass configuration..."
|
|
"${PYTHON_BIN}" - "${CONFIG_PATH}" <<'PY'
|
|
import sys
|
|
import subprocess
|
|
import json
|
|
import configparser
|
|
|
|
def die(msg):
|
|
print(f"ERROR: {msg}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
c = configparser.ConfigParser()
|
|
c.read(sys.argv[1])
|
|
g = c["Global"] if "Global" in c else {}
|
|
e = c["env_setup"] if "env_setup" in c else {}
|
|
i = c["init_cluster"] if "init_cluster" in c else {}
|
|
|
|
ctx = g.get("APP_CLUSTER_KUBECONTEXT") or e.get("APP_CLUSTER_KUBECONTEXT") or i.get("app_cluster_kubecontext", "")
|
|
if not ctx:
|
|
sys.exit(0)
|
|
|
|
# Resolve StorageClass name
|
|
sc_name = g.get("SUPABASE_STORAGE_CLASS") or c.get("Supabase", "STORAGE_CLASS", fallback="supabase-standard")
|
|
|
|
try:
|
|
out = subprocess.check_output(["kubectl", "--context", ctx, "get", "storageclass", sc_name, "-o", "json"], stderr=subprocess.DEVNULL, text=True)
|
|
sc = json.loads(out)
|
|
provisioner = sc.get("provisioner", "")
|
|
params = sc.get("parameters", {})
|
|
disk_type = params.get("type", "")
|
|
|
|
if provisioner != "pd.csi.storage.gke.io":
|
|
die(f"storage-class configuration error: StorageClass '{sc_name}' must use provisioner 'pd.csi.storage.gke.io', but uses '{provisioner}'.")
|
|
|
|
if disk_type != "pd-standard":
|
|
die(f"storage-class configuration error: StorageClass '{sc_name}' must use parameters.type 'pd-standard', but uses '{disk_type}'.")
|
|
|
|
print(f"Verified live Supabase StorageClass '{sc_name}' (type: {disk_type}).", file=sys.stderr)
|
|
except subprocess.CalledProcessError:
|
|
die(f"storage-class configuration error: Required StorageClass '{sc_name}' does not exist on cluster '{ctx}'.")
|
|
PY
|
|
fi
|
|
|
|
"${PYTHON_BIN}" -m prole.deploy_pipeline --config "${CONFIG_PATH}" "$@"
|
|
|
|
# Post-deploy summary
|
|
echo ""
|
|
echo "Public Endpoints:"
|
|
ROOT_DIR="${ROOT_DIR}" "${PYTHON_BIN}" - "${CONFIG_PATH}" <<'PY'
|
|
import sys
|
|
import os
|
|
import subprocess
|
|
import json
|
|
import configparser
|
|
import time
|
|
import re
|
|
|
|
def get_config(path):
|
|
c = configparser.ConfigParser()
|
|
c.read(path)
|
|
# Check [Global] then [env_setup] then [init_cluster] for contexts
|
|
g = c["Global"] if "Global" in c else {}
|
|
e = c["env_setup"] if "env_setup" in c else {}
|
|
i = c["init_cluster"] if "init_cluster" in c else {}
|
|
|
|
app_ctx = g.get("APP_CLUSTER_KUBECONTEXT") or e.get("APP_CLUSTER_KUBECONTEXT") or i.get("app_cluster_kubecontext", "")
|
|
db_ctx = g.get("DB_CLUSTER_KUBECONTEXT") or e.get("DB_CLUSTER_KUBECONTEXT") or i.get("db_cluster_kubecontext", "")
|
|
# GitLab namespace usually defaults to 'gitlab'
|
|
gitlab_ns = g.get("GITLAB_NAMESPACE") or e.get("GITLAB_NAMESPACE") or i.get("gitlab_namespace", "gitlab")
|
|
return app_ctx, db_ctx, gitlab_ns
|
|
|
|
def get_ip(host, ctx):
|
|
if not ctx: return "pending"
|
|
# Try Ingress first
|
|
try:
|
|
out = subprocess.check_output(["kubectl", "--context", ctx, "get", "ingress", "-A", "-o", "json"], stderr=subprocess.DEVNULL, text=True)
|
|
items = json.loads(out).get("items", [])
|
|
for item in items:
|
|
for rule in item.get("spec", {}).get("rules", []):
|
|
if rule.get("host") == host:
|
|
ing = item.get("status", {}).get("loadBalancer", {}).get("ingress", [])
|
|
if ing:
|
|
val = ing[0].get("ip") or ing[0].get("hostname")
|
|
if val: return val
|
|
except: pass
|
|
|
|
return "pending"
|
|
|
|
def get_ingress_status(ctx, namespace, ingress_name):
|
|
status = {
|
|
"name": ingress_name,
|
|
"exists": False,
|
|
"address": "pending",
|
|
"hosts": [],
|
|
}
|
|
if not ctx:
|
|
return status
|
|
try:
|
|
out = subprocess.check_output([
|
|
"kubectl", "--context", ctx, "get", "ingress", ingress_name,
|
|
"-n", namespace, "-o", "json"
|
|
], stderr=subprocess.DEVNULL, text=True)
|
|
item = json.loads(out)
|
|
status["exists"] = True
|
|
status["hosts"] = [
|
|
rule.get("host")
|
|
for rule in item.get("spec", {}).get("rules", [])
|
|
if rule.get("host")
|
|
]
|
|
ing = item.get("status", {}).get("loadBalancer", {}).get("ingress", [])
|
|
if ing:
|
|
addr = ing[0].get("ip") or ing[0].get("hostname")
|
|
if addr:
|
|
status["address"] = addr
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
return status
|
|
|
|
def check_supabase_pvcs(ctx):
|
|
if not ctx: return None
|
|
try:
|
|
from datetime import datetime, timezone
|
|
out = subprocess.check_output(["kubectl", "--context", ctx, "get", "events", "-n", "supabase", "-o", "json"], stderr=subprocess.DEVNULL, text=True)
|
|
events = json.loads(out).get("items", [])
|
|
|
|
# Fetch current PVC states to ignore Bound ones
|
|
out_pvc = subprocess.check_output(["kubectl", "--context", ctx, "get", "pvc", "-n", "supabase", "-o", "json"], stderr=subprocess.DEVNULL, text=True)
|
|
pvcs = json.loads(out_pvc).get("items", [])
|
|
pvc_status = {p.get("metadata", {}).get("name"): p.get("status", {}).get("phase") for p in pvcs}
|
|
|
|
blocking_pvcs = []
|
|
for event in events:
|
|
obj = event.get("involvedObject", {})
|
|
if obj.get("kind") == "PersistentVolumeClaim":
|
|
pvc_name = obj.get("name")
|
|
|
|
# Skip Bound PVCs
|
|
if pvc_status.get(pvc_name) == "Bound":
|
|
continue
|
|
|
|
# Skip old events (> 10 mins)
|
|
last_ts = event.get("lastTimestamp") or event.get("eventTime")
|
|
if last_ts:
|
|
try:
|
|
ts = datetime.fromisoformat(last_ts.replace('Z', '+00:00'))
|
|
if (datetime.now(timezone.utc) - ts).total_seconds() > 600:
|
|
continue
|
|
except: pass
|
|
|
|
msg = event.get("message", "")
|
|
reason = event.get("reason", "")
|
|
if "QUOTA_EXCEEDED" in msg or "ProvisioningFailed" in reason:
|
|
if "exceeded" in msg.lower() or "quota" in msg.lower():
|
|
blocking_pvcs.append({
|
|
"pvc": pvc_name,
|
|
"message": msg,
|
|
})
|
|
|
|
stale_retained = []
|
|
out_pv = subprocess.check_output([
|
|
"kubectl", "--context", ctx, "get", "pv", "-o", "json"
|
|
], stderr=subprocess.DEVNULL, text=True)
|
|
pvs = json.loads(out_pv).get("items", [])
|
|
for pv in pvs:
|
|
spec = pv.get("spec", {})
|
|
status = pv.get("status", {})
|
|
claim = spec.get("claimRef") or {}
|
|
claim_ns = claim.get("namespace")
|
|
claim_name = claim.get("name")
|
|
if claim_ns != "supabase":
|
|
continue
|
|
if not str(claim_name or "").startswith("supabase-"):
|
|
continue
|
|
|
|
phase = status.get("phase", "")
|
|
reclaim = spec.get("persistentVolumeReclaimPolicy", "")
|
|
if reclaim != "Retain" or phase not in ("Released", "Failed"):
|
|
continue
|
|
|
|
csi = spec.get("csi") or {}
|
|
gce = spec.get("gcePersistentDisk") or {}
|
|
labels = (pv.get("metadata") or {}).get("labels") or {}
|
|
disk_handle = csi.get("volumeHandle") or gce.get("pdName") or "<unknown>"
|
|
zone = (
|
|
labels.get("topology.kubernetes.io/zone")
|
|
or labels.get("failure-domain.beta.kubernetes.io/zone")
|
|
or (csi.get("volumeAttributes") or {}).get("topology.gke.io/zone")
|
|
or ""
|
|
)
|
|
if not zone and isinstance(disk_handle, str):
|
|
m = re.search(r"/zones/([^/]+)/disks/[^/]+$", disk_handle)
|
|
if m:
|
|
zone = m.group(1)
|
|
stale_retained.append({
|
|
"pv": (pv.get("metadata") or {}).get("name", "?"),
|
|
"pvc": claim_name or "?",
|
|
"phase": phase or "Unknown",
|
|
"storageClass": spec.get("storageClassName") or "<unset>",
|
|
"diskHandle": disk_handle,
|
|
"zone": zone or "<unknown>",
|
|
})
|
|
|
|
if blocking_pvcs or stale_retained:
|
|
return {
|
|
"blockingPVCs": blocking_pvcs,
|
|
"staleRetainedPVs": stale_retained,
|
|
}
|
|
except: pass
|
|
return None
|
|
|
|
def reconcile_gitlab(app_ctx, db_ctx, ns, config_path):
|
|
if not app_ctx: return
|
|
print(f"\nReconciling GitLab workloads in namespace '{ns}'...")
|
|
print(f" Using config: {config_path}")
|
|
print(f" Using APP context: {app_ctx}")
|
|
if db_ctx:
|
|
print(f" Using DB context: {db_ctx}")
|
|
|
|
try:
|
|
# Get all deployments in one go
|
|
out = subprocess.check_output(["kubectl", "--context", app_ctx, "-n", ns, "get", "deployments", "-o", "json"], text=True)
|
|
deploys = json.loads(out).get("items", [])
|
|
|
|
# Get all replicasets in one go
|
|
out = subprocess.check_output(["kubectl", "--context", app_ctx, "-n", ns, "get", "replicasets", "-o", "json"], text=True)
|
|
all_rss = json.loads(out).get("items", [])
|
|
except Exception as e:
|
|
print(f" Warning: Could not fetch GitLab workloads for reconciliation: {e}")
|
|
return
|
|
|
|
# Targets specifically mentioned in requirements or operator managed
|
|
gitlab_workloads = [
|
|
"gitlab-sidekiq-all-in-1-v2",
|
|
"gitlab-gitlab-shell",
|
|
"gitlab-kas",
|
|
"gitlab-registry",
|
|
"gitlab-webservice-default"
|
|
]
|
|
|
|
summary = []
|
|
over_deployed_detected = False
|
|
|
|
for dep in deploys:
|
|
name = dep['metadata']['name']
|
|
# Filter for operator-managed GitLab workloads only
|
|
is_gitlab = False
|
|
if name in gitlab_workloads:
|
|
is_gitlab = True
|
|
elif name.startswith("gitlab-") and not any(x in name for x in ["-migration", "-secrets", "-minio-create-buckets", "-db-config"]):
|
|
is_gitlab = True
|
|
|
|
if not is_gitlab:
|
|
continue
|
|
|
|
spec = dep.get('spec', {})
|
|
status = dep.get('status', {})
|
|
|
|
# Target should be 1 for these specific GitLab workloads (Requirement 11)
|
|
if name in gitlab_workloads:
|
|
desired = 1
|
|
else:
|
|
desired = spec.get('replicas', 0)
|
|
|
|
live = status.get('replicas', 0)
|
|
updated = status.get('updatedReplicas', 0)
|
|
|
|
# Enforce desired replica count (Requirement 3 & 4)
|
|
# Reconciliation compares:
|
|
# - desired replicas (from Helm values)
|
|
# - actual replicas (from Deployment.spec.replicas)
|
|
actual_spec_replicas = spec.get('replicas', 0)
|
|
if name in gitlab_workloads and actual_spec_replicas != desired:
|
|
print(f" [FIX] Deployment '{name}' spec.replicas ({actual_spec_replicas}) differs from desired ({desired}).")
|
|
over_deployed_detected = True
|
|
|
|
summary.append({
|
|
"name": name,
|
|
"desired": desired,
|
|
"live": live,
|
|
"updated": updated
|
|
})
|
|
|
|
# Check for Gitaly storage class drift on GKE (Requirement 3)
|
|
is_gke = "gke" in sys.argv[1].lower()
|
|
if is_gke:
|
|
try:
|
|
out = subprocess.check_output(["kubectl", "--context", app_ctx, "-n", ns, "get", "gitlab", "-o", "json"], text=True)
|
|
crs = json.loads(out).get("items", [])
|
|
for cr in crs:
|
|
spec = cr.get("spec", {})
|
|
live_sc = spec.get("chart", {}).get("values", {}).get("global", {}).get("persistence", {}).get("storageClass")
|
|
if not live_sc:
|
|
live_sc = spec.get("chart", {}).get("values", {}).get("gitlab", {}).get("gitaly", {}).get("persistence", {}).get("storageClass")
|
|
|
|
# Desired is 'standard' in this environment by default
|
|
desired_sc = "standard"
|
|
|
|
if live_sc:
|
|
# Normalize comparison: standard and standard-rwo are equivalent on GKE
|
|
gke_standard_variants = ["standard", "standard-rwo"]
|
|
if live_sc in gke_standard_variants and desired_sc in gke_standard_variants:
|
|
continue # Equivalent
|
|
|
|
if live_sc != desired_sc:
|
|
print(f" [FIX] GitLab CR '{cr['metadata']['name']}' storageClass '{live_sc}' differs from desired '{desired_sc}'.")
|
|
over_deployed_detected = True
|
|
except:
|
|
pass
|
|
|
|
# Print summary (Requirement 8)
|
|
print(f"\n{'GitLab Deployment':<35} {'Desired':<8} {'Live':<8} {'Updated':<8}")
|
|
print("-" * 70)
|
|
for s in summary:
|
|
print(f"{s['name']:<35} {s['desired']:<8} {s['live']:<8} {s['updated']:<8}")
|
|
|
|
if over_deployed_detected:
|
|
print("\nEnforcing desired configuration by re-applying GitLab Helm release/CR...")
|
|
# Re-apply the Helm release (Requirement 2 & 4)
|
|
try:
|
|
root_dir = os.environ.get("ROOT_DIR", ".")
|
|
env = os.environ.copy()
|
|
# Explicitly pass contexts and config path to avoid fallbacks
|
|
env["APP_CLUSTER_KUBECONTEXT"] = app_ctx
|
|
if db_ctx:
|
|
env["DB_CLUSTER_KUBECONTEXT"] = db_ctx
|
|
env["KUBECTL_CONTEXT"] = app_ctx
|
|
env["PROLE_DEPLOY_CFG"] = config_path
|
|
|
|
subprocess.run(["/usr/bin/env", "bash", f"{root_dir}/etc/init_gitlab.sh", "--config", config_path, "deploy"], env=env, check=True)
|
|
except Exception as e:
|
|
print(f" ERROR: Failed to re-apply GitLab configuration: {e}")
|
|
sys.exit(1)
|
|
|
|
# Final Convergence Check (Requirement 6)
|
|
still_over_deployed = False
|
|
for s in summary:
|
|
# If we detected drift, we re-applied the configuration.
|
|
# Now we check if it converged.
|
|
if over_deployed_detected:
|
|
# Re-fetch live status for the final check
|
|
try:
|
|
out = subprocess.check_output(["kubectl", "--context", app_ctx, "-n", ns, "get", "deployment", s['name'], "-o", "json"], text=True)
|
|
d = json.loads(out)
|
|
st = d.get('status', {})
|
|
# Compare against the intended 'desired' count (1), not necessarily the (potentially stale) spec.replicas
|
|
if st.get('replicas', 0) > s['desired'] or st.get('updatedReplicas', 0) != s['desired']:
|
|
print(f"ERROR: GitLab deployment '{s['name']}' failed to converge (live={st.get('replicas')}, desired={s['desired']}).")
|
|
still_over_deployed = True
|
|
except:
|
|
pass
|
|
elif s['live'] > s['desired'] or s['updated'] != s['desired']:
|
|
print(f"ERROR: GitLab deployment '{s['name']}' is not converged.")
|
|
still_over_deployed = True
|
|
|
|
if still_over_deployed:
|
|
sys.exit(1)
|
|
|
|
app_ctx, db_ctx, gitlab_ns = get_config(sys.argv[1])
|
|
config_path = sys.argv[1]
|
|
|
|
# Run GitLab reconciliation first
|
|
reconcile_gitlab(app_ctx, db_ctx, gitlab_ns, config_path)
|
|
|
|
# Supabase frontdoor ingresses that must converge together
|
|
required_supabase_ingresses = [
|
|
{"name": "supabase-kong", "endpoint": "api.0.knoe.dev"},
|
|
{"name": "supabase-studio", "endpoint": "db.0.knoe.dev"},
|
|
]
|
|
timeout = 300 # 5 minutes
|
|
start = time.time()
|
|
supabase_ingress_state = {}
|
|
|
|
while time.time() - start < timeout:
|
|
pvc_state = check_supabase_pvcs(app_ctx)
|
|
if pvc_state:
|
|
stale = pvc_state.get("staleRetainedPVs") or []
|
|
if stale:
|
|
print("\nERROR: Supabase PVC provisioning is blocked by stale retained PV/GCE disk artifacts from previous runs.")
|
|
for item in stale:
|
|
print(f" - PV: {item['pv']}")
|
|
print(f" PVC: {item['pvc']}")
|
|
print(f" Phase/Reclaim: {item['phase']}/{item.get('storageClass','<unset>')} Retain")
|
|
print(f" Disk: {item['diskHandle']} (zone: {item['zone']})")
|
|
print("\nAction: stale Supabase Retain PVs should be auto-cleaned during deploy; delete the remaining backing GCE disks listed above, or enable SUPABASE_AUTO_CLEAN_RETAINED_GCE_DISKS=true with gcloud/project env configured.")
|
|
sys.exit(1)
|
|
|
|
blockers = pvc_state.get("blockingPVCs") or []
|
|
if blockers:
|
|
print("\nERROR: Supabase PVC provisioning failed (Storage Quota Exceeded).")
|
|
for item in blockers:
|
|
print(f" - PVC: {item.get('pvc')}")
|
|
print(f" GKE Error: {item.get('message')}")
|
|
sys.exit(1)
|
|
|
|
pending = False
|
|
supabase_ingress_state = {}
|
|
for req in required_supabase_ingresses:
|
|
ing = get_ingress_status(app_ctx, "supabase", req["name"])
|
|
supabase_ingress_state[req["name"]] = ing
|
|
if not ing.get("exists") or ing.get("address") == "pending":
|
|
pending = True
|
|
break
|
|
if not pending:
|
|
break
|
|
time.sleep(10)
|
|
|
|
if time.time() - start >= timeout:
|
|
print("\nERROR: Timeout waiting for Supabase ingress reconciliation (both supabase-kong and supabase-studio are required).")
|
|
for req in required_supabase_ingresses:
|
|
ing = supabase_ingress_state.get(req["name"]) or get_ingress_status(app_ctx, "supabase", req["name"])
|
|
print(f" - {req['name']} ({req['endpoint']}): exists={ing.get('exists')} address={ing.get('address')} hosts={','.join(ing.get('hosts') or []) or '-'}")
|
|
print("Ingress Controller Diagnosis (kubectl get events -n supabase):")
|
|
try:
|
|
subprocess.run(["kubectl", "--context", app_ctx, "get", "events", "-n", "supabase", "--sort-by=.lastTimestamp"], check=False)
|
|
print("\nIngress Resource Status:")
|
|
subprocess.run(["kubectl", "--context", app_ctx, "describe", "ingress", "supabase-kong", "-n", "supabase"], check=False)
|
|
subprocess.run(["kubectl", "--context", app_ctx, "describe", "ingress", "supabase-studio", "-n", "supabase"], check=False)
|
|
except:
|
|
pass
|
|
sys.exit(1)
|
|
|
|
kong_state = supabase_ingress_state.get("supabase-kong") or get_ingress_status(app_ctx, "supabase", "supabase-kong")
|
|
studio_state = supabase_ingress_state.get("supabase-studio") or get_ingress_status(app_ctx, "supabase", "supabase-studio")
|
|
if (not kong_state.get("exists") or kong_state.get("address") == "pending" or
|
|
not studio_state.get("exists") or studio_state.get("address") == "pending"):
|
|
print("\nERROR: Supabase frontdoor is not converged: both supabase-kong and supabase-studio ingresses must exist and have addresses.")
|
|
print(f" - supabase-kong (api.0.knoe.dev): exists={kong_state.get('exists')} address={kong_state.get('address')}")
|
|
print(f" - supabase-studio (db.0.knoe.dev): exists={studio_state.get('exists')} address={studio_state.get('address')}")
|
|
sys.exit(1)
|
|
|
|
endpoints = [
|
|
("api.0.knoe.dev", "Supabase Kong", "APP", kong_state.get("address") or "pending"),
|
|
("db.0.knoe.dev", "Supabase Studio", "APP", studio_state.get("address") or "pending"),
|
|
("api.prole.org", "knoe-svc-kong", "APP", get_ip("api.prole.org", app_ctx)),
|
|
("git.prole.org", "GitLab", "APP", get_ip("git.prole.org", app_ctx)),
|
|
("svc.prole.org", "Grafana", "APP", get_ip("svc.prole.org", app_ctx)),
|
|
]
|
|
|
|
print(f"\n{'hostname':<20} {'service':<20} {'cluster':<10} {'external IP':<15}")
|
|
print("-" * 70)
|
|
for host, svc, clus, ip in endpoints:
|
|
print(f"{host:<20} {svc:<20} {clus:<10} {ip:<15}")
|
|
PY
|