"""Cluster lifecycle management (k3d / k3s / k8s).""" import os import re import subprocess import sys import threading import time import tkinter as tk from pathlib import Path from tkinter import ttk, messagebox, filedialog import json from knoe import prole_conf from knoe import screen as ui from knoe.config import ( _collect_cfg_vars_from_data, _encrypt_cfg_secret, _expand_path_expr, _merge_kubeconfig, _parse_gcp_cfg, _write_k3s_kubeconfig, ) from knoe.core.actions import _reset_k3s_namespace from knoe.core.env import ( PROJECT_ROOT, _default_opentofu_pipeline_url, _deployment_mode_from_env, _deployment_target_label, _find_kubeconfig_file, _k3d_prole_data_volume_args, _normalize_cluster_env, _normalize_k3s_token, _resolve_k3s_connection as _resolve_k3s_connection_fn, _safe_str, ) from knoe.core.prod_config import ProdConfigApi, build_config, to_api_payload class ClusterScreenMixin: """Cluster lifecycle management (k3d / k3s / k8s).""" def _render_init_cluster_page(self): # Letterhead at top right content_width = self.bg_canvas.winfo_width() or 975 right_margin = content_width - 48 ui.canvas_text( self, right_margin, 40, "knoe.dev", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "infrastructure.auto()", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) # Tighten vertical spacing on this screen so the main status area (traffic lights + # console output) has enough room in the fixed-size window. self._render_title("Cluster Environment", y=110) self._render_paragraph( "Select the cluster environment for this session." " Dev runs a local k3d cluster on this machine." " Service and Prod connect to remote clusters using the server URL and token below.", y=155, ) # Cluster Selection (Radio Buttons) x_label = 48 y = 210 self._canvas_items.append( ui.canvas_text( self, x_label, y, "Select Environment:", fill="black", font=("SF Pro Text", 14, "bold"), ) ) y += 36 cluster_options = [("dev", "Dev"), ("service", "Service"), ("prod", "Prod")] # We need to trace cluster_env if not already traced if not hasattr(self, "_cluster_env_trace"): self._cluster_env_trace = self.cluster_env.trace_add( "write", self._on_cluster_env_change ) rb_x = x_label + 20 rb_gap = 150 for idx, (val, name) in enumerate(cluster_options): rb = tk.Radiobutton( self.bg_canvas, text=name, variable=self.cluster_env, value=val, bg="white", fg="black", activebackground="white", selectcolor="white", font=("SF Pro Text", 11), ) rb_window = self.bg_canvas.create_window( rb_x + idx * rb_gap, y, window=rb, anchor="nw" ) self._canvas_items.append(rb_window) self._overlay_widgets.append(rb) selected_env_key = self._cluster_env_key() y += 36 if selected_env_key == "dev": # Combined Select / Add Cluster widget self._canvas_items.append( ui.canvas_text( self, x_label, y, "Cluster:", fill="black", font=("SF Pro Text", 12, "bold"), ) ) y += 28 clusters = self._get_k3d_cluster_list() combo = ttk.Combobox( self.bg_canvas, textvariable=self.selected_k3d_cluster, values=clusters, width=30, ) if not self.selected_k3d_cluster.get() and clusters: self.selected_k3d_cluster.set(clusters[0]) combo.bind( "<>", lambda _evt: self._on_k3d_cluster_select() ) combo_win = self.bg_canvas.create_window( x_label + 20, y, window=combo, anchor="nw" ) self._canvas_items.append(combo_win) self._overlay_widgets.append(combo) add_btn = tk.Button( self.bg_canvas, text="Add", command=self._on_create_k3d_cluster, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", activeforeground="black", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, ) add_win = self.bg_canvas.create_window( x_label + 290, y - 4, window=add_btn, anchor="nw" ) self._canvas_items.append(add_win) self._overlay_widgets.append(add_btn) delete_btn = tk.Button( self.bg_canvas, text="Delete", command=self._on_delete_k3d_cluster, bg="#F5F5DC", fg="#cc0000", activebackground="#E5E5D5", activeforeground="#cc0000", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, ) delete_win = self.bg_canvas.create_window( x_label + 350, y - 4, window=delete_btn, anchor="nw" ) self._canvas_items.append(delete_win) self._overlay_widgets.append(delete_btn) y += 40 else: # Service / Prod Section # Context selection for both Service and Prod (managed via kubectx) self._canvas_items.append( ui.canvas_text( self, x_label, y, "Kubernetes Context:", fill="black", font=("SF Pro Text", 12, "bold"), ) ) values = self._get_kubectx_list() combo = ttk.Combobox( self.bg_canvas, textvariable=self.selected_kubectx, values=values, state="readonly", width=40, ) combo.bind("<>", self._on_kubectx_select) if not self.selected_kubectx.get() and values: # Default to the current context if it exists in the list; otherwise pick # a sensible default without switching immediately. current_ctx = (getattr(self, "_kubectx_applied", "") or "").strip() if not current_ctx: current_ctx = (self._current_kubectl_context() or "").strip() if current_ctx: self._kubectx_applied = current_ctx if current_ctx and current_ctx in values: self.selected_kubectx.set(current_ctx) elif selected_env_key == "service" and "prole-k3s" in values: self.selected_kubectx.set("prole-k3s") else: self.selected_kubectx.set(values[0]) combo_win = self.bg_canvas.create_window( x_label + 270, y - 6, window=combo, anchor="nw" ) self._canvas_items.append(combo_win) self._overlay_widgets.append(combo) # Keep references for UI tests / layout verification. self._kubectx_combo = combo self._kubectx_combo_canvas_window = combo_win ui.canvas_text( self, x_label + 580, y + 2, "via kubectx", fill="#6e6e73", font=("SF Pro Text", 10), ) apply_btn = tk.Button( self.bg_canvas, text="Apply", command=self._on_kubectx_apply, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, state="disabled", ) apply_win = self.bg_canvas.create_window( x_label + 670, y - 10, window=apply_btn, anchor="nw" ) self._canvas_items.append(apply_win) self._overlay_widgets.append(apply_btn) # Keep references for UI tests / layout verification. self._kubectx_apply_btn = apply_btn self._kubectx_apply_btn_canvas_window = apply_win self._update_kubectx_apply_button() y += 34 if selected_env_key == "prod": # Staging directory input self._canvas_items.append( ui.canvas_text( self, x_label, y, "Local Artifact Staging Directory:", fill="black", font=("SF Pro Text", 12, "bold"), ) ) entry = tk.Entry( self.bg_canvas, textvariable=self.prod_artifacts_path, width=50, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) entry_win = self.bg_canvas.create_window( x_label + 260, y - 6, window=entry, anchor="nw" ) self._canvas_items.append(entry_win) self._overlay_widgets.append(entry) browse_btn = tk.Button( self.bg_canvas, text="Browse...", command=lambda: self.prod_artifacts_path.set( filedialog.askdirectory() or self.prod_artifacts_path.get() ), ) browse_win = self.bg_canvas.create_window( x_label + 770, y - 8, window=browse_btn, anchor="nw" ) self._canvas_items.append(browse_win) self._overlay_widgets.append(browse_btn) y += 34 ui.canvas_text( self, x_label + 20, y, "(Used by etc/deploy_pipeline.sh --mode gcp)", fill="#6e6e73", font=("SF Pro Text", 10), ) y += 40 self._ensure_prod_config_state() self._canvas_items.append( ui.canvas_text( self, x_label, y, "Production Configuration", fill="black", font=("SF Pro Text", 12, "bold"), ) ) y += 24 notebook = ttk.Notebook(self.bg_canvas) cloud_tab = tk.Frame(notebook, bg="white") db_tab = tk.Frame(notebook, bg="white") backup_tab = tk.Frame(notebook, bg="white") auth_tab = tk.Frame(notebook, bg="white") migration_tab = tk.Frame(notebook, bg="white") plan_tab = tk.Frame(notebook, bg="white") notebook.add(cloud_tab, text="Cloud") notebook.add(db_tab, text="Database") notebook.add(backup_tab, text="Backup + Storage") notebook.add(auth_tab, text="Auth + Routing") notebook.add(migration_tab, text="Migration") notebook.add(plan_tab, text="Plan / Apply") for frame in [cloud_tab, db_tab, backup_tab, auth_tab, migration_tab, plan_tab]: frame.columnconfigure(1, weight=1) def _add_row( frame: tk.Frame, row: int, label: str, key: str, width: int = 40, readonly: bool = False, ) -> int: tk.Label( frame, text=label, bg="white", fg="black", font=("SF Pro Text", 10), anchor="w", ).grid(row=row, column=0, sticky="w", padx=8, pady=3) entry = tk.Entry( frame, textvariable=self.prod_form_vars[key], width=width, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 10), ) if readonly: entry.configure(state="readonly") entry.grid(row=row, column=1, sticky="ew", padx=8, pady=3) return row + 1 cloud_row = 0 cloud_row = _add_row(cloud_tab, cloud_row, "Provider", "cloud.provider", readonly=True) cloud_row = _add_row(cloud_tab, cloud_row, "GCP Project ID", "cloud.projectId") cloud_row = _add_row(cloud_tab, cloud_row, "Region", "cloud.region") cloud_row = _add_row(cloud_tab, cloud_row, "Cluster", "cloud.clusterName") cloud_row = _add_row(cloud_tab, cloud_row, "VPC Mode", "cloud.vpcMode") cloud_row = _add_row(cloud_tab, cloud_row, "VPC Name", "cloud.vpcName") cloud_row = _add_row(cloud_tab, cloud_row, "Subnet Name", "cloud.subnetName") cloud_row = _add_row( cloud_tab, cloud_row, "Artifact Registry", "cloud.artifactRegistry" ) _add_row(cloud_tab, cloud_row, "DNS Zone", "cloud.dnsZone") def _on_load_gcp_cfg(): gcp_cfg_path = self._resolve_prole_conf_dir() / "prod" / "gcp.cfg" loaded = _parse_gcp_cfg(gcp_cfg_path) if not loaded: import tkinter.messagebox as _mb _mb.showwarning( "GCP Setup", f"No values found at:\n{gcp_cfg_path}\n\n" "Run: python3 etc/config.py --mode k8s --provider gcp", ) return upper = {k.upper(): v for k, v in loaded.items()} if upper.get("PROJECT_ID"): self.prod_form_vars["cloud.projectId"].set(upper["PROJECT_ID"]) if upper.get("REGION"): self.prod_form_vars["cloud.region"].set(upper["REGION"]) self.prole_cfg_data.setdefault("GCP", {}).update(upper) self._gcp_cfg_values = dict(self.prole_cfg_data["GCP"]) load_gcp_btn = tk.Button( cloud_tab, text="Load from gcp.cfg", command=_on_load_gcp_cfg, font=("SF Pro Text", 10), ) load_gcp_btn.grid( row=cloud_row + 1, column=0, columnspan=2, sticky="w", padx=8, pady=(6, 2) ) db_row = 0 db_row = _add_row( db_tab, db_row, "Namespace", "kubernetes.namespace", readonly=True ) db_row = _add_row(db_tab, db_row, "CNPG Cluster", "database.clusterName") db_row = _add_row(db_tab, db_row, "Postgres Version", "database.postgresVersion") db_row = _add_row(db_tab, db_row, "Instances", "database.instances") db_row = _add_row(db_tab, db_row, "Storage Class", "database.storageClass") db_row = _add_row(db_tab, db_row, "Storage Size (Gi)", "database.storageSizeGi") db_row = _add_row(db_tab, db_row, "App Database", "database.appDatabase") db_row = _add_row(db_tab, db_row, "Meta Database", "database.metaDatabase") db_row = _add_row(db_tab, db_row, "App User", "database.appUser") _add_row(db_tab, db_row, "Admin User", "database.adminUser") backup_row = 0 backup_row = _add_row( backup_tab, backup_row, "Backup Bucket", "backups.backupBucket" ) backup_row = _add_row(backup_tab, backup_row, "WAL Bucket", "backups.walBucket") _add_row(backup_tab, backup_row, "Retention Days", "backups.retentionDays") auth_row = 0 auth_row = _add_row(auth_tab, auth_row, "Auth Provider", "auth.provider", readonly=True) auth_row = _add_row(auth_tab, auth_row, "Issuer", "auth.issuer") auth_row = _add_row(auth_tab, auth_row, "OIDC Client ID Ref", "auth.clientId") auth_row = _add_row(auth_tab, auth_row, "OIDC Client Secret Ref", "auth.clientSecret") auth_row = _add_row( auth_tab, auth_row, "Bootstrap Admin Email", "auth.bootstrapAdminEmail", ) auth_row = _add_row(auth_tab, auth_row, "Frontdoor Host", "routing.frontdoorHost") auth_row = _add_row(auth_tab, auth_row, "Platform Domain", "routing.platformDomain") _add_row(auth_tab, auth_row, "TLS Mode", "routing.tlsMode") migration_row = 0 migration_row = _add_row( migration_tab, migration_row, "Source Environment", "migration.sourceEnvironment", ) migration_row = _add_row(migration_tab, migration_row, "Mode", "migration.mode") migration_row = _add_row( migration_tab, migration_row, "Source Host", "migration.sourceHost" ) migration_row = _add_row( migration_tab, migration_row, "Source Port", "migration.sourcePort" ) migration_row = _add_row( migration_tab, migration_row, "Source Database", "migration.sourceDatabase", ) migration_row = _add_row( migration_tab, migration_row, "Source User", "migration.sourceUser" ) migration_row = _add_row( migration_tab, migration_row, "Source Password Ref", "migration.sourcePasswordRef", ) tk.Checkbutton( migration_tab, text="Continuous until cutover", variable=self.prod_bool_vars["migration.continuousUntilCutover"], bg="white", fg="black", activebackground="white", selectcolor="white", font=("SF Pro Text", 10), ).grid( row=migration_row, column=0, columnspan=2, sticky="w", padx=8, pady=6, ) btn_row = tk.Frame(plan_tab, bg="white") btn_row.grid(row=0, column=0, columnspan=2, sticky="w", padx=8, pady=4) save_btn = tk.Button( btn_row, text="Save Config", command=lambda: self._prod_put_config(show_dialog=True), ) plan_btn = tk.Button(btn_row, text="Plan", command=self._on_prod_plan) apply_cfg_btn = tk.Button(btn_row, text="Apply", command=self._on_prod_apply) refresh_btn = tk.Button( btn_row, text="Refresh Status", command=self._on_prod_refresh_status ) for idx, btn in enumerate([save_btn, plan_btn, apply_cfg_btn, refresh_btn]): btn.grid(row=0, column=idx, sticky="w", padx=(0, 6)) tk.Label( plan_tab, textvariable=self.prod_apply_status, bg="white", fg="#1c1c1e", anchor="w", font=("SF Pro Text", 10, "bold"), ).grid(row=1, column=0, columnspan=2, sticky="ew", padx=8, pady=2) preview_text = tk.Text( plan_tab, height=8, width=95, bg="#f8f8f8", fg="black", wrap="word", font=("SF Mono", 9), ) preview_text.grid(row=2, column=0, columnspan=2, sticky="nsew", padx=8, pady=4) tk.Label( plan_tab, text="Apply Progress Logs", bg="white", fg="black", anchor="w", font=("SF Pro Text", 10, "bold"), ).grid(row=3, column=0, columnspan=2, sticky="w", padx=8, pady=(0, 2)) logs_text = tk.Text( plan_tab, height=4, width=95, bg="#111111", fg="#9de79f", wrap="word", font=("SF Mono", 9), ) logs_text.grid(row=4, column=0, columnspan=2, sticky="nsew", padx=8, pady=2) plan_tab.rowconfigure(2, weight=1) self._prod_preview_text_widget = preview_text self._prod_logs_text_widget = logs_text notebook_win = self.bg_canvas.create_window( x_label + 20, y, window=notebook, anchor="nw", width=890, height=290, ) self._canvas_items.append(notebook_win) self._overlay_widgets.append(notebook) y += 304 self._prod_push_preview_widgets() if not self.prod_apply_status.get().strip(): self.prod_apply_status.set("idle/ready: Waiting for plan") ui.canvas_text( self, x_label, y, "Common Services are configured on the next screen.", fill="#6e6e73", font=("SF Pro Text", 11), ) def _render_common_services_page(self): # Letterhead at top right content_width = self.bg_canvas.winfo_width() or 975 right_margin = content_width - 48 ui.canvas_text( self, right_margin, 40, "knoe.dev", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "infrastructure.auto()", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) self._render_title("Common Services", y=110) self._render_paragraph( "Evaluate, deploy, or repair common services." " All service traffic lights must be green before continuing.", y=155, ) selected_env_key = self._cluster_env_key() has_cluster = True if selected_env_key == "dev": has_cluster = bool((self.selected_k3d_cluster.get() or "").strip()) x_label = 48 y = 220 if not has_cluster: ui.canvas_text( self, x_label, y, "Select or create a Dev cluster before configuring common services.", fill="#6e6e73", font=("SF Pro Text", 12), ) self._common_services_success = False self._service_traffic_lights = {} self.update_footer() return ui.canvas_text( self, x_label, y, "Common Core Services Namespace:", fill="black", font=("SF Pro Text", 12, "bold"), ) ns_entry = tk.Entry( self.bg_canvas, textvariable=self.service_namespace, width=20, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) ns_win = self.bg_canvas.create_window( x_label + 270, y - 6, window=ns_entry, anchor="nw" ) self._canvas_items.append(ns_win) self._overlay_widgets.append(ns_entry) self._service_namespace_entry = ns_entry self._service_namespace_entry_canvas_window = ns_win save_btn = tk.Button( self.bg_canvas, text="Save", command=self._validate_and_save_cluster_config, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, ) save_win = self.bg_canvas.create_window( x_label + 480, y - 10, window=save_btn, anchor="nw" ) self._canvas_items.append(save_win) self._overlay_widgets.append(save_btn) eval_btn = tk.Button( self.bg_canvas, text="Evaluate", command=self._verify_k3s_services, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, ) eval_win = self.bg_canvas.create_window( x_label + 560, y - 10, window=eval_btn, anchor="nw" ) self._canvas_items.append(eval_win) self._overlay_widgets.append(eval_btn) deploy_btn = tk.Button( self.bg_canvas, text="Deploy", command=self._deploy_k3s_services, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, ) deploy_win = self.bg_canvas.create_window( x_label + 660, y - 10, window=deploy_btn, anchor="nw" ) self._canvas_items.append(deploy_win) self._overlay_widgets.append(deploy_btn) self._k3s_deploy_btn = deploy_btn repair_btn = tk.Button( self.bg_canvas, text="Repair", command=self._deploy_k3s_services, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, ) repair_win = self.bg_canvas.create_window( x_label + 758, y - 10, window=repair_btn, anchor="nw" ) self._canvas_items.append(repair_win) self._overlay_widgets.append(repair_btn) y += 42 components = [ ("registry", "Registry"), ("certmgr", "Cert Manager"), ("garage", "Garage"), ("kong", "Kong"), ("openbao", "OpenBao"), ("opentofu", "OpenTofu"), ] lights = {} for idx, (key, label) in enumerate(components): row = idx // 3 col = idx % 3 lx = x_label + 20 + (col * 290) ly = y + 8 + (row * 36) indicator = self.bg_canvas.create_oval( lx, ly, lx + 14, ly + 14, fill="#ff9f0a", outline="#8e8e93", width=1, ) self._canvas_items.append(indicator) ui.canvas_text( self, lx + 22, ly + 7, label, fill="black", font=("SF Pro Text", 10), anchor="w", ) lights[key] = indicator self._service_traffic_lights = lights y += 90 notebook = ttk.Notebook(self.bg_canvas) status_tab = tk.Frame(notebook, bg="white") deploy_tab = tk.Frame(notebook, bg="white") notebook.add(status_tab, text="Evaluate") notebook.add(deploy_tab, text="Deploy / Repair") status_console = ui.TerminalConsole(status_tab, highlightthickness=0, bd=0) status_console.pack(fill="both", expand=True, padx=4, pady=4) deploy_console = ui.TerminalConsole(deploy_tab, highlightthickness=0, bd=0) deploy_console.pack(fill="both", expand=True, padx=4, pady=4) notebook_win = self.bg_canvas.create_window( x_label, y, window=notebook, anchor="nw", width=890, height=430, ) self._canvas_items.append(notebook_win) self._overlay_widgets.extend( [notebook, status_tab, deploy_tab, status_console, deploy_console] ) self._k3s_service_notebook = notebook self._k3s_service_status_tab = status_tab self._k3s_service_deploy_tab = deploy_tab self._k3s_service_status_console = status_console self._k3s_service_deploy_console = deploy_console self._common_services_success = bool( getattr(self, "_common_services_success", False) or getattr(self, "_all_services_green", False) ) self.update_footer() self._verify_k3s_services() def _cluster_env_key(self, env_label: str | None = None) -> str: """Return normalized cluster env key ('dev', 'service', 'prod').""" value = env_label if value is None: try: value = self.cluster_env.get() except Exception: value = "" try: if hasattr(value, "get") and callable(value.get): value = value.get() except Exception: pass if not value: try: value = ( (self.prole_cfg_data.get("Initialize Cluster", {}) or {}) .get("ENVIRONMENT", "") .strip() ) except Exception: value = "" if not value: try: value = ( (self.prole_cfg_data.get("Global", {}) or {}) .get("CLUSTER_ENV", "") .strip() ) except Exception: value = "" key = _normalize_cluster_env(value) if not key: return "dev" return key def _is_valid_k8s_namespace(self, value: str | None) -> bool: ns = _safe_str(value).strip() if not ns or len(ns) > 63: return False return re.fullmatch(r"[a-z0-9]([-a-z0-9]*[a-z0-9])?", ns) is not None def _get_service_namespace(self) -> str: """Resolve service namespace with UI/env/config fallbacks. Safety rule: avoid defaulting to the Kubernetes `default` namespace unless we truly cannot infer a better option. """ # 1) Explicit user/config/env choice wins. candidate_namespaces: list[str] = [] try: candidate_namespaces.append((self.service_namespace.get() or "").strip()) except Exception: pass try: candidate_namespaces.append( ( (self.prole_cfg_data.get("Global", {}) or {}) .get("SERVICE_NAMESPACE", "") .strip() ) ) except Exception: pass candidate_namespaces.append((os.environ.get("SERVICE_NAMESPACE") or "").strip()) for ns in candidate_namespaces: if self._is_valid_k8s_namespace(ns): return ns # 2) If the cluster is reachable, infer the common services namespace. inferred = "" try: inferred = (self._infer_common_services_namespace() or "").strip() except Exception: inferred = "" if self._is_valid_k8s_namespace(inferred): return inferred # 3) Stable defaults for known environments. try: # For local dev (k3d) and service (k3s), keep knoe-system stable across restarts. if self._cluster_env_key() in ("dev", "service"): return "knoe-system" except Exception: pass # 4) Absolute last resort. return "default" def _infer_common_services_namespace(self) -> str: """Infer the namespace hosting common services. If the cluster is reachable, try to locate an existing Docker Registry workload (image containing `registry:2`). If found, treat that namespace as the common services namespace. This method is read-only. """ # Cache to avoid repeated `kubectl -A` probes during periodic refresh. cache_ttl_s = 60 now = time.time() cached_ns = getattr(self, "_common_services_ns_cache", "") cached_at = getattr(self, "_common_services_ns_cache_at", 0.0) if cached_ns and (now - float(cached_at or 0.0)) < cache_ttl_s: return cached_ns # Ensure kubeconfig is available when possible (read-only action). try: self._ensure_k3s_kubeconfig_merged() except Exception: pass # Build a kubectl command suitable for the active mode. try: mode = self._deployment_mode() except Exception: mode = None try: base_cmd = self._kubectl_base_cmd(mode) except Exception: base_cmd = ["kubectl"] # Respect an existing kubeconfig if present. env = os.environ.copy() try: kc = _find_kubeconfig_file(env) except Exception: kc = _find_kubeconfig_file() if kc: env["KUBECONFIG"] = kc def _probe(kind: str) -> list[str]: cmd = base_cmd + ["get", kind, "-A", "-o", "json"] res = subprocess.run( cmd, capture_output=True, text=True, env=env, timeout=5, ) if res.returncode != 0: return [] try: data = json.loads(res.stdout or "{}") except Exception: return [] out = [] for item in (data.get("items") or []): try: ns = (item.get("metadata") or {}).get("namespace") or "" containers = ( (((item.get("spec") or {}).get("template") or {}).get("spec") or {}).get( "containers" ) or [] ) for c in containers: img = (c.get("image") or "").strip() if "registry:2" in img: if ns: out.append(ns) break except Exception: continue return out namespaces = [] # Deployments are the most common shape; fall back to StatefulSets. namespaces.extend(_probe("deploy")) if not namespaces: namespaces.extend(_probe("sts")) # Prefer knoe-system if present, otherwise prefer a non-default namespace. ns_choice = "" if namespaces: if "knoe-system" in namespaces: ns_choice = "knoe-system" else: non_default = [n for n in namespaces if n and n != "default"] ns_choice = (non_default[0] if non_default else namespaces[0]) or "" if ns_choice: setattr(self, "_common_services_ns_cache", ns_choice) setattr(self, "_common_services_ns_cache_at", now) return ns_choice def _on_cluster_env_change(self, *args): # Keep configuration environment selection explicit and stable. try: conf_dir = self._resolve_prole_conf_dir() prole_conf.activate_environment(conf_dir, self.cluster_env.get()) except Exception: pass self._set_deploy_target_from_cluster_env() # Trigger status check or refresh UI env_key = self._cluster_env_key() if env_key == "service": try: self._apply_k3s_defaults() except Exception: pass # Prefer prole-k3s context if present, but require explicit Apply. try: values = self._get_kubectx_list() except Exception: values = [] if hasattr(self, "selected_kubectx") and "prole-k3s" in (values or []): self.selected_kubectx.set("prole-k3s") self._update_kubectx_apply_button() elif env_key == "dev": name = self.selected_k3d_cluster.get() if name: if hasattr(self, "selected_kubectx"): self.selected_kubectx.set(f"k3d-{name}") self._switch_kubectx(f"k3d-{name}") self._verify_k3s_services() self.show_page("init_cluster") def _deployment_mode(self) -> str: try: env_val = self.cluster_env.get() except Exception: env_val = "" return _deployment_mode_from_env(env_val) def _deployment_pipeline_url(self, target_key: str) -> str: env_url = ( os.environ.get("PROLE_OPENTOFU_URL") or os.environ.get("OPENTOFU_URL") or "" ).strip() if env_url: return env_url section = None if target_key == "service": section = "Service Cluster (k3s)" elif target_key == "prod": section = "Prod Cluster (k8s)" if section: url = ( (self.prole_cfg_data.get(section, {}) or {}) .get("PIPELINE_URL", "") .strip() ) if url: return url return _default_opentofu_pipeline_url() def _set_deploy_target_from_cluster_env(self): if not hasattr(self, "deploy_target"): return if getattr(self, "_syncing_deploy_target", False): return self._syncing_deploy_target = True try: self.deploy_target.set(_deployment_target_label(self.cluster_env.get())) except Exception: pass finally: self._syncing_deploy_target = False def _on_deploy_target_change(self, *args): if getattr(self, "_syncing_deploy_target", False): return target = "" try: target = self.deploy_target.get() except Exception: target = "" key = _normalize_cluster_env(target) if key == "dev": desired = "dev" elif key == "service": desired = "service" elif key == "prod": desired = "prod" else: return self._syncing_deploy_target = True try: if self.cluster_env.get() != desired: self.cluster_env.set(desired) finally: self._syncing_deploy_target = False def _dashboard_kong_status(self, env: str | None = None) -> dict: """Check Kubernetes Dashboard (Kong) pod health and return status info.""" ns = "kubernetes-dashboard" base_cmd = self._kubectl_base_cmd(env) def _pod_healthy(pod_status: dict) -> bool: phase = (pod_status.get("phase") or "").strip() if phase not in ("Running", "Succeeded"): return False statuses = pod_status.get("containerStatuses") or [] if not statuses: return phase == "Succeeded" for cs in statuses: if cs.get("ready") is True: continue state = cs.get("state") or {} waiting = state.get("waiting") or {} reason = (waiting.get("reason") or "").lower() if reason: return False return False return True # Ensure namespace exists rc_ns, out_ns = self._run_cmd_capture( base_cmd + ["get", "ns", ns, "-o", "name"], timeout=6 ) if rc_ns != 0 or ns not in (out_ns or ""): return { "ok": False, "msg": "Dashboard (Kong): Missing", "fill": "#ff9f0a", "reset_pods": [], } rc, out = self._run_cmd_capture( base_cmd + ["get", "pods", "-n", ns, "-o", "json"], timeout=8 ) if rc != 0: return { "ok": False, "msg": "Dashboard (Kong): Unable to query", "fill": "#ff9f0a", "reset_pods": [], } try: payload = json.loads(out or "{}") except Exception: return { "ok": False, "msg": "Dashboard (Kong): Unable to parse status", "fill": "#ff9f0a", "reset_pods": [], } kong_pods = [] for item in payload.get("items", []): name = (item.get("metadata", {}) or {}).get("name", "") name_lc = name.lower() if "kong" in name_lc and "dashboard" in name_lc: kong_pods.append(item) if not kong_pods: return { "ok": False, "msg": "Dashboard (Kong): Missing", "fill": "#ff9f0a", "reset_pods": [], } unhealthy = [] for item in kong_pods: name = (item.get("metadata", {}) or {}).get("name", "") status = item.get("status") or {} if not _pod_healthy(status): unhealthy.append(name) if unhealthy: return { "ok": False, "msg": "Dashboard (Kong): Unhealthy (safe to reset)", "fill": "#ff9f0a", "reset_pods": unhealthy, } return { "ok": True, "msg": "Dashboard (Kong): Running", "fill": "#34c759", "reset_pods": [], } def _set_cluster_env_message( self, text: str, color: str = "#34c759", clear_after_ms: int | None = None ): """Set a transient status message on the Cluster Environment screen.""" def _apply(): if not hasattr(self, "_cluster_env_message_label"): return if not self.bg_canvas.winfo_exists(): return try: self.bg_canvas.itemconfig( self._cluster_env_message_label, text=text, fill=color ) except Exception: pass self.safe_after(_apply) if clear_after_ms is not None and clear_after_ms >= 0: def _clear(): if not hasattr(self, "_cluster_env_message_label"): return if not self.bg_canvas.winfo_exists(): return try: self.bg_canvas.itemconfig(self._cluster_env_message_label, text="") except Exception: pass self.safe_after(_clear, delay=clear_after_ms) def _authority_context_missing(self) -> bool: """Detect missing authority Docker context when Kerberos is enabled.""" try: if not self.kerberos_enabled.get(): return False except Exception: return False candidates = [] env_home = (os.environ.get("PROLE_HOME") or "").strip() if env_home: candidates.append(Path(env_home)) candidates.append(PROJECT_ROOT) for base in candidates: try: if (base / "authority").is_dir(): return False if (base / "prole" / "authority").is_dir(): return False except Exception: continue return True def _maybe_run_repair_pipeline(self, status: dict, kong_status: dict | None): """Run repair pipeline when cluster is ready and anomalies are detected.""" if not status.get("cluster_ok"): return anomalies = [] if kong_status and not kong_status.get("ok", True): anomalies.append("dashboard") if self._authority_context_missing(): anomalies.append("authority") if not anomalies: return now = time.time() if self._repair_inflight: return if now - self._repair_last_run_at < self._repair_cooldown_s: return self._repair_last_run_at = now self._repair_inflight = True def worker(): try: env_key = status.get("env") or self._cluster_env_key() if env_key == "service": ns = self._get_service_namespace() else: ns = (self.db_namespace.get() or "").strip() or "default" env = self._script_env_for_namespace(ns) if env_key == "service": env["PROLE_MODE"] = "k3s" elif env_key == "dev": env["PROLE_MODE"] = "k3d" self._set_cluster_env_message( "Repair pipeline started", "#34c759", clear_after_ms=3000 ) self.controller.run_script( "repair_pipeline.sh", args=["-n", ns], env=env, ) self._set_cluster_env_message( "Repair pipeline complete", "#34c759", clear_after_ms=3000 ) except Exception as e: self._set_cluster_env_message( f"Repair pipeline failed: {e}", "#ff3b30", clear_after_ms=6000 ) finally: self._repair_inflight = False threading.Thread(target=worker, daemon=True).start() def _check_opentofu_health(self) -> bool: """Check OpenTofu readiness in the selected namespace.""" try: env_key = self._cluster_env_key() except Exception: env_key = "dev" try: if env_key == "service": ns = self._get_service_namespace() else: ns = (self.db_namespace.get() or "").strip() or "default" except Exception: ns = "default" try: base_cmd = self._kubectl_base_cmd(env_key) res_dep = subprocess.run( base_cmd + [ "-n", ns, "get", "deploy", "opentofu", "-o", "jsonpath={.status.readyReplicas}", ], capture_output=True, text=True, timeout=3, ) ready_dep = res_dep.returncode == 0 and ( res_dep.stdout or "0" ).strip() not in ("", "0") if ready_dep: return True res_sts = subprocess.run( base_cmd + [ "-n", ns, "get", "statefulset", "opentofu", "-o", "jsonpath={.status.readyReplicas}", ], capture_output=True, text=True, timeout=3, ) ready_sts = res_sts.returncode == 0 and ( res_sts.stdout or "0" ).strip() not in ("", "0") return ready_sts except Exception: return False def _check_k8s_cluster(self, env_label: str) -> tuple[bool, str]: env_key = self._cluster_env_key(env_label) label = ( "K3s Cluster" if env_key == "service" else "Prod Cluster" if env_key == "prod" else "Kubernetes Cluster" ) try: kubectl = subprocess.run(["which", "kubectl"], capture_output=True) if kubectl.returncode != 0: return False, f"{label}: kubectl not found" if env_key == "service": kubeconfig = _find_kubeconfig_file() if not kubeconfig: server, token = self._k3s_connection_info() if not server: return False, f"{label}: missing server URL or kubeconfig" cmd = self._kubectl_base_cmd(env_key) + ["cluster-info"] res = subprocess.run(cmd, capture_output=True, text=True, timeout=8) ok = res.returncode == 0 except Exception: ok = False msg = f"{label}: Connected" if ok else f"{label}: Not reachable" return ok, msg def _get_k3d_cluster_list(self) -> list[str]: """Get list of k3d clusters.""" try: res = subprocess.run( ["k3d", "cluster", "list", "--no-headers"], capture_output=True, text=True, timeout=5, ) if res.returncode == 0: # Extract first column (cluster names) return [ line.split()[0] for line in res.stdout.splitlines() if line.strip() ] except Exception: pass return [] def _on_create_k3d_cluster(self): name = self.selected_k3d_cluster.get().strip() if not name: messagebox.showerror("Error", "Cluster name cannot be empty") return # Persist the typed name so the refreshed page keeps it selected self.k3d_cluster_name.set(name) # If the cluster already exists just select it (triggers notebook display) existing = self._get_k3d_cluster_list() if name in existing: self.selected_k3d_cluster.set(name) self.show_page("init_cluster") return # Refresh the page so the notebook/console is visible before streaming output self.selected_k3d_cluster.set(name) self.show_page("init_cluster") def do_create(): console = getattr(self, "_k3s_service_deploy_console", None) if console: console.clear() console.write( f"Creating k3d cluster '{name}' @ {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n" ) try: if self._k3s_service_notebook and self._k3s_service_deploy_tab: self._k3s_service_notebook.select(self._k3s_service_deploy_tab) except Exception: pass try: # Build the k3d create command prole_data = str(self._resolve_env_dir("PROLE_DATA", "data")) volume_args = _k3d_prole_data_volume_args(prole_data) reg_args = [] if getattr(self, "registry_url", None): if self.registry_url.startswith("localhost:5000"): reg_args = ["--registry-create", "prole-registry:0.0.0.0:5000"] else: reg_args = ["--registry-use", self.registry_url] cmd = ( ["k3d", "cluster", "create", name, "-a", "2", "--wait"] + volume_args + reg_args + ["--timestamps"] ) if console: console.write(f"$ {' '.join(cmd)}\n\n") proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=PROJECT_ROOT, ) for line in proc.stdout: if console: console.write(line) proc.wait() if proc.returncode != 0: if console: console.write( f"\nk3d cluster create failed (exit {proc.returncode})\n" ) return if console: console.write(f"\nCluster '{name}' created successfully.\n\n") # Merge kubeconfig so kubectl can reach the new cluster try: subprocess.run( [ "k3d", "kubeconfig", "merge", name, "--kubeconfig-switch-context", ], capture_output=True, text=True, timeout=10, ) if console: console.write(f"Kubeconfig merged for cluster '{name}'.\n\n") except Exception as kce: if console: console.write(f"Warning: failed to merge kubeconfig: {kce}\n") # Refresh UI and then deploy common services check self.safe_after(self._refresh_k3d_ui) # Trigger common services check in the deploy console self.safe_after(self._deploy_k3s_services) except Exception as e: if console: console.write(f"\nError: {e}\n") self.safe_after( lambda: messagebox.showerror( "Error", f"Failed to create k3d cluster: {e}" ) ) threading.Thread(target=do_create, daemon=True).start() def _refresh_k3d_ui(self): clusters = self._get_k3d_cluster_list() self.k3d_cluster_list.set(clusters) name = self.k3d_cluster_name.get().strip() if name in clusters: self.selected_k3d_cluster.set(name) self.show_page("init_cluster") def _on_k3d_cluster_select(self, *args): """Called when a k3d cluster is selected from the dropdown.""" cluster_name = self.selected_k3d_cluster.get() if cluster_name: try: # Merge kubeconfig and switch context to the selected cluster subprocess.run( ["k3d", "kubeconfig", "merge", cluster_name, "--switch-context"], capture_output=True, text=True, ) # Ensure kubectx also knows about it self._switch_kubectx(f"k3d-{cluster_name}") except Exception: pass self.show_page("init_cluster") def _on_delete_k3d_cluster(self): """Delete the selected k3d cluster after user confirmation.""" name = self.selected_k3d_cluster.get().strip() if not name: messagebox.showerror("Error", "No cluster selected to delete.") return confirmed = messagebox.askyesno( "Confirm Delete", f"Are you sure you want to delete cluster '{name}'?" ) if not confirmed: return # Show the notebook/console before streaming output self.show_page("init_cluster") def do_delete(): console = getattr(self, "_k3s_service_deploy_console", None) if console: console.clear() console.write( f"Deleting k3d cluster '{name}' @ {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n" ) try: if self._k3s_service_notebook and self._k3s_service_deploy_tab: self._k3s_service_notebook.select(self._k3s_service_deploy_tab) except Exception: pass try: cmd = ["k3d", "cluster", "delete", name] if console: console.write(f"$ {' '.join(cmd)}\n\n") proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=PROJECT_ROOT, ) for line in proc.stdout: if console: console.write(line) proc.wait() if proc.returncode != 0: if console: console.write( f"\nk3d cluster delete failed (exit {proc.returncode})\n" ) return if console: console.write(f"\nCluster '{name}' deleted successfully.\n") # Clear selection and refresh UI self.selected_k3d_cluster.set("") self.k3d_cluster_name.set("") self.safe_after(self._refresh_k3d_ui) except Exception as e: if console: console.write(f"\nError: {e}\n") self.safe_after( lambda: messagebox.showerror( "Error", f"Failed to delete k3d cluster: {e}" ) ) threading.Thread(target=do_delete, daemon=True).start() def _on_kubectx_select(self, *args): """Called when a generic kubernetes context is selected from the dropdown.""" self._update_kubectx_apply_button() ctx = (self.selected_kubectx.get() or "").strip() if ctx: self._set_cluster_env_message( f"Context change pending: {ctx} (click Apply)", "#ff9500", clear_after_ms=4000, ) def _on_kubectx_apply(self): ctx = (self.selected_kubectx.get() or "").strip() if not ctx: return ok = self._switch_kubectx(ctx) if ok: self._kubectx_applied = ctx self._update_kubectx_apply_button() self._set_cluster_env_message( f"Context applied: {ctx}", "#34c759", clear_after_ms=3000 ) self._verify_k3s_services() self.show_page("init_cluster") def _update_kubectx_apply_button(self): btn = getattr(self, "_kubectx_apply_btn", None) if not btn: return selected = "" try: if hasattr(self, "selected_kubectx") and self.selected_kubectx is not None: selected = (self.selected_kubectx.get() or "").strip() except Exception: selected = "" applied = (getattr(self, "_kubectx_applied", "") or "").strip() state = "normal" if (selected and selected != applied) else "disabled" try: btn.config(state=state) except Exception: pass def _switch_kubectx(self, context_name: str) -> bool: """Switch the current kubernetes context using kubectx or kubectl.""" if not context_name: return False # Determine which kubeconfig to modify. # Use the same logic as the rest of the application. kc = _find_kubeconfig_file() env = os.environ.copy() if kc: env["KUBECONFIG"] = kc try: # Try kubectx first res = subprocess.run( ["kubectx", context_name], capture_output=True, text=True, env=env ) ok = res.returncode == 0 err = (res.stderr or "").strip() if not ok: # Fallback to kubectl config use-context res2 = subprocess.run( ["kubectl", "config", "use-context", context_name], capture_output=True, text=True, env=env ) ok = res2.returncode == 0 if not ok: err2 = (res2.stderr or "").strip() err = err2 or err # Update the UI dropdown to match (but don't trigger re-select) if ( ok and hasattr(self, "selected_kubectx") and self.selected_kubectx.get() != context_name ): try: # Check if it's in the current values list current_values = [] if hasattr(self, "kubectx_list") and self.kubectx_list is not None: current_values = list(self.kubectx_list.get() or []) if not current_values: current_values = self._get_kubectx_list() if context_name in (current_values or []): self.selected_kubectx.set(context_name) except Exception: pass if not ok: try: messagebox.showerror( "Kubernetes Context", f"Failed to switch to context '{context_name}'.\n\n{err or 'Unknown error'}", ) except Exception: pass return ok except Exception as e: try: messagebox.showerror( "Kubernetes Context", f"Failed to switch to context '{context_name}': {e}", ) except Exception: pass return False def _current_kubectl_context(self) -> str: kc = _find_kubeconfig_file() env = os.environ.copy() if kc: env["KUBECONFIG"] = kc try: res = subprocess.run( ["kubectl", "config", "current-context"], capture_output=True, text=True, env=env, ) if res.returncode == 0: return (res.stdout or "").strip() except Exception: pass return "" def _get_kubectx_list(self) -> list[str]: """Get list of kubernetes contexts.""" kc = _find_kubeconfig_file() env = os.environ.copy() if kc: env["KUBECONFIG"] = kc try: res = subprocess.run(["kubectx"], capture_output=True, text=True, env=env) if res.returncode == 0: return res.stdout.strip().split("\n") # Fallback to kubectl res = subprocess.run( ["kubectl", "config", "get-contexts", "-o", "name"], capture_output=True, text=True, env=env ) if res.returncode == 0: return res.stdout.strip().split("\n") except Exception: pass return ["default"] def _ensure_prod_config_state(self): if getattr(self, "_prod_config_state_ready", False): return self._prod_config_state_ready = True self.prod_config_api = getattr(self, "prod_config_api", ProdConfigApi()) self.prod_form_vars = { "metadata.name": tk.StringVar(value="knoey-root"), "metadata.environment": tk.StringVar(value="production"), "cloud.provider": tk.StringVar(value="gcp"), "cloud.projectId": tk.StringVar(value=""), "cloud.region": tk.StringVar(value="us-central1"), "cloud.clusterName": tk.StringVar(value="knoe-prod"), "cloud.vpcMode": tk.StringVar(value="managed"), "cloud.vpcName": tk.StringVar(value=""), "cloud.subnetName": tk.StringVar(value=""), "cloud.artifactRegistry": tk.StringVar(value=""), "cloud.dnsZone": tk.StringVar(value="knoe-dev-zone"), "kubernetes.namespace": tk.StringVar(value="ecosystem-0"), "database.clusterName": tk.StringVar(value="knoe-db"), "database.postgresVersion": tk.StringVar(value="16"), "database.instances": tk.StringVar(value="3"), "database.storageClass": tk.StringVar(value="premium-rwo"), "database.storageSizeGi": tk.StringVar(value="100"), "database.appDatabase": tk.StringVar(value="knoey"), "database.metaDatabase": tk.StringVar(value="knoe_meta"), "database.appUser": tk.StringVar(value="knoey_app"), "database.adminUser": tk.StringVar(value="knoe_admin"), "backups.backupBucket": tk.StringVar(value="knoe-0-backups"), "backups.walBucket": tk.StringVar(value="knoe-0-wal"), "backups.retentionDays": tk.StringVar(value="14"), "auth.provider": tk.StringVar(value="google-workspace-oidc"), "auth.issuer": tk.StringVar(value="https://accounts.google.com"), "auth.clientId": tk.StringVar(value="secretref://google-oidc-client-id"), "auth.clientSecret": tk.StringVar( value="secretref://google-oidc-client-secret" ), "auth.bootstrapAdminEmail": tk.StringVar(value="admin@knoey.com"), "routing.frontdoorHost": tk.StringVar(value="knoey.com"), "routing.platformDomain": tk.StringVar(value="knoe.dev"), "routing.tlsMode": tk.StringVar(value="managed"), "migration.sourceEnvironment": tk.StringVar(value="prole.org"), "migration.mode": tk.StringVar(value="snapshot-restore"), "migration.sourceHost": tk.StringVar(value="knoe-local-db.prole.org"), "migration.sourcePort": tk.StringVar(value="5432"), "migration.sourceDatabase": tk.StringVar(value="knoey"), "migration.sourceUser": tk.StringVar(value="replication_user"), "migration.sourcePasswordRef": tk.StringVar( value="secretref://local-source-db-password" ), } self.prod_bool_vars = { "migration.continuousUntilCutover": tk.BooleanVar(value=False) } self.prod_tab_messages = tk.StringVar(value="") self.prod_apply_status = tk.StringVar(value="") self.prod_preview_yaml = "" self.prod_preview_tf_vars = "" self.prod_preview_install_plan = "" self.prod_apply_logs = "" self._prod_preview_text_widget = None self._prod_logs_text_widget = None gcp = getattr(self, "_gcp_cfg_values", {}) or {} if gcp.get("PROJECT_ID"): self.prod_form_vars["cloud.projectId"].set(gcp["PROJECT_ID"]) if gcp.get("REGION") and self.prod_form_vars["cloud.region"].get() in ("", "us-central1"): self.prod_form_vars["cloud.region"].set(gcp["REGION"]) try: saved_yaml = ( (getattr(self, "prole_cfg_data", None) or {}) .get("Prod Cluster (k8s)", {}) .get("PRODUCTION_CONFIG_YAML", "") .strip() ) if saved_yaml: import yaml as _yaml saved_doc = _yaml.safe_load(saved_yaml) or {} restored = build_config(saved_doc) self.prod_config_api.put_prod_config(to_api_payload(restored)) payload = self.prod_config_api.get_prod_config() self._prod_apply_payload_to_vars(payload) except Exception: pass def _prod_apply_payload_to_vars(self, payload: dict | None): cfg = build_config(payload or {}) values = { "metadata.name": cfg.metadata.name, "metadata.environment": cfg.metadata.environment, "cloud.provider": cfg.cloud.provider, "cloud.projectId": cfg.cloud.projectId, "cloud.region": cfg.cloud.region, "cloud.clusterName": cfg.cloud.clusterName, "cloud.vpcMode": cfg.cloud.vpcMode, "cloud.vpcName": cfg.cloud.vpcName or "", "cloud.subnetName": cfg.cloud.subnetName or "", "cloud.artifactRegistry": cfg.cloud.artifactRegistry, "cloud.dnsZone": cfg.cloud.dnsZone, "kubernetes.namespace": cfg.kubernetes.namespace, "database.clusterName": cfg.database.clusterName, "database.postgresVersion": cfg.database.postgresVersion, "database.instances": str(cfg.database.instances), "database.storageClass": cfg.database.storageClass, "database.storageSizeGi": str(cfg.database.storageSizeGi), "database.appDatabase": cfg.database.appDatabase, "database.metaDatabase": cfg.database.metaDatabase, "database.appUser": cfg.database.appUser, "database.adminUser": cfg.database.adminUser, "backups.backupBucket": cfg.backups.backupBucket, "backups.walBucket": cfg.backups.walBucket, "backups.retentionDays": str(cfg.backups.retentionDays), "auth.provider": cfg.auth.provider, "auth.issuer": cfg.auth.issuer, "auth.clientId": cfg.auth.clientId, "auth.clientSecret": cfg.auth.clientSecret, "auth.bootstrapAdminEmail": cfg.auth.bootstrapAdminEmail, "routing.frontdoorHost": cfg.routing.frontdoorHost, "routing.platformDomain": cfg.routing.platformDomain, "routing.tlsMode": cfg.routing.tlsMode, "migration.sourceEnvironment": cfg.migration.sourceEnvironment, "migration.mode": cfg.migration.mode, "migration.sourceHost": cfg.migration.sourceHost, "migration.sourcePort": str(cfg.migration.sourcePort), "migration.sourceDatabase": cfg.migration.sourceDatabase, "migration.sourceUser": cfg.migration.sourceUser, "migration.sourcePasswordRef": cfg.migration.sourcePasswordRef, } for key, value in values.items(): if key in self.prod_form_vars: self.prod_form_vars[key].set(value) self.prod_bool_vars["migration.continuousUntilCutover"].set( bool(cfg.migration.continuousUntilCutover) ) def _prod_payload_from_vars(self) -> dict: self._ensure_prod_config_state() def _to_int(key: str) -> int: try: return int((self.prod_form_vars[key].get() or "").strip()) except Exception: return 0 doc = { "kind": "KnoeProductionConfig", "metadata": { "ecosystemId": 0, "name": self.prod_form_vars["metadata.name"].get().strip(), "environment": self.prod_form_vars["metadata.environment"].get().strip(), }, "spec": { "cloud": { "provider": self.prod_form_vars["cloud.provider"].get().strip(), "projectId": self.prod_form_vars["cloud.projectId"].get().strip(), "region": self.prod_form_vars["cloud.region"].get().strip(), "clusterName": self.prod_form_vars["cloud.clusterName"].get().strip(), "vpcMode": self.prod_form_vars["cloud.vpcMode"].get().strip(), "vpcName": self.prod_form_vars["cloud.vpcName"].get().strip() or None, "subnetName": self.prod_form_vars["cloud.subnetName"].get().strip() or None, "artifactRegistry": self.prod_form_vars[ "cloud.artifactRegistry" ].get().strip(), "dnsZone": self.prod_form_vars["cloud.dnsZone"].get().strip(), }, "kubernetes": { "namespace": self.prod_form_vars["kubernetes.namespace"].get().strip() }, "database": { "clusterName": self.prod_form_vars["database.clusterName"].get().strip(), "postgresVersion": self.prod_form_vars[ "database.postgresVersion" ].get().strip(), "instances": _to_int("database.instances"), "storageClass": self.prod_form_vars["database.storageClass"].get().strip(), "storageSizeGi": _to_int("database.storageSizeGi"), "appDatabase": self.prod_form_vars["database.appDatabase"].get().strip(), "metaDatabase": self.prod_form_vars["database.metaDatabase"].get().strip(), "appUser": self.prod_form_vars["database.appUser"].get().strip(), "adminUser": self.prod_form_vars["database.adminUser"].get().strip(), }, "backups": { "backupBucket": self.prod_form_vars["backups.backupBucket"].get().strip(), "walBucket": self.prod_form_vars["backups.walBucket"].get().strip(), "retentionDays": _to_int("backups.retentionDays"), }, "auth": { "provider": self.prod_form_vars["auth.provider"].get().strip(), "issuer": self.prod_form_vars["auth.issuer"].get().strip(), "clientId": self.prod_form_vars["auth.clientId"].get().strip(), "clientSecret": self.prod_form_vars["auth.clientSecret"].get().strip(), "bootstrapAdminEmail": self.prod_form_vars[ "auth.bootstrapAdminEmail" ].get().strip(), }, "routing": { "frontdoorHost": self.prod_form_vars[ "routing.frontdoorHost" ].get().strip(), "platformDomain": self.prod_form_vars[ "routing.platformDomain" ].get().strip(), "tlsMode": self.prod_form_vars["routing.tlsMode"].get().strip(), }, "migration": { "sourceEnvironment": self.prod_form_vars[ "migration.sourceEnvironment" ].get().strip(), "mode": self.prod_form_vars["migration.mode"].get().strip(), "sourceHost": self.prod_form_vars["migration.sourceHost"].get().strip(), "sourcePort": _to_int("migration.sourcePort"), "sourceDatabase": self.prod_form_vars[ "migration.sourceDatabase" ].get().strip(), "sourceUser": self.prod_form_vars["migration.sourceUser"].get().strip(), "sourcePasswordRef": self.prod_form_vars[ "migration.sourcePasswordRef" ].get().strip(), "continuousUntilCutover": self.prod_bool_vars[ "migration.continuousUntilCutover" ].get(), }, }, } return to_api_payload(build_config(doc)) def _prod_format_messages(self, errors: list[str], warnings: list[str]) -> str: lines = [] if errors: lines.append("Errors:") lines.extend([f"- {line}" for line in errors]) if warnings: if lines: lines.append("") lines.append("Warnings:") lines.extend([f"- {line}" for line in warnings]) return "\n".join(lines) def _prod_push_preview_widgets(self): combined = [ "=== Canonical YAML Config Preview ===", self.prod_preview_yaml or "(no plan yet)", "=== OpenTofu Variables Preview ===", self.prod_preview_tf_vars or "(no plan yet)", "=== install.py Module Plan ===", self.prod_preview_install_plan or "(no plan yet)", "=== Warnings / Errors ===", self.prod_tab_messages.get() or "(none)", ] preview_text = "\n".join(combined).strip() + "\n" widget = getattr(self, "_prod_preview_text_widget", None) if widget is not None: try: widget.configure(state="normal") widget.delete("1.0", tk.END) widget.insert("1.0", preview_text) widget.configure(state="disabled") except Exception: pass log_widget = getattr(self, "_prod_logs_text_widget", None) if log_widget is not None: try: log_widget.configure(state="normal") log_widget.delete("1.0", tk.END) log_widget.insert("1.0", (self.prod_apply_logs or "") + "\n") log_widget.configure(state="disabled") except Exception: pass def _prod_sync_status_and_logs(self): self._ensure_prod_config_state() try: status = self.prod_config_api.get_status() or {} except Exception: status = {} try: logs = (self.prod_config_api.get_logs() or {}).get("logs", []) except Exception: logs = [] phase = (status.get("phase") or "unknown").strip() state = (status.get("state") or "unknown").strip() message = (status.get("message") or "").strip() self.prod_apply_status.set(f"{phase}/{state}: {message}".strip()) self.prod_apply_logs = "\n".join(logs[-300:]) self._prod_push_preview_widgets() def _prod_put_config(self, show_dialog: bool = False) -> dict: self._ensure_prod_config_state() result = self.prod_config_api.put_prod_config(self._prod_payload_from_vars()) cfg_doc = result.get("config") or {} if cfg_doc: self._prod_apply_payload_to_vars(cfg_doc) errors = list(result.get("errors") or []) warnings = list(result.get("warnings") or []) self.prod_tab_messages.set(self._prod_format_messages(errors, warnings)) self._prod_push_preview_widgets() if show_dialog: if errors: messagebox.showwarning("Production Config", "\n".join(errors)) else: note = "Production config saved." if warnings: note += "\n\nWarnings:\n" + "\n".join(warnings) messagebox.showinfo("Production Config", note) return result def _on_prod_plan(self): self._ensure_prod_config_state() put_result = self._prod_put_config(show_dialog=False) if put_result.get("errors"): self._prod_sync_status_and_logs() return result = self.prod_config_api.post_plan(self._prod_payload_from_vars()) self.prod_preview_yaml = result.get("yaml", "") self.prod_preview_tf_vars = json.dumps( result.get("opentofuVars") or {}, indent=2, sort_keys=True ) self.prod_preview_install_plan = "\n".join( f"- {line}" for line in (result.get("installPlan") or []) ) self.prod_tab_messages.set( self._prod_format_messages( list(result.get("errors") or []), list(result.get("warnings") or []) ) ) self._prod_sync_status_and_logs() def _on_prod_apply(self): self._ensure_prod_config_state() put_result = self._prod_put_config(show_dialog=False) if put_result.get("errors"): self._prod_sync_status_and_logs() return result = self.prod_config_api.post_apply(self._prod_payload_from_vars()) self.prod_preview_yaml = result.get("yaml", "") self.prod_preview_tf_vars = json.dumps( result.get("opentofuVars") or {}, indent=2, sort_keys=True ) self.prod_preview_install_plan = "\n".join( f"- {line}" for line in (result.get("installPlan") or []) ) self.prod_tab_messages.set( self._prod_format_messages( list(result.get("errors") or []), list(result.get("warnings") or []) ) ) self._prod_sync_status_and_logs() def _on_prod_refresh_status(self): self._ensure_prod_config_state() self._prod_sync_status_and_logs() def _validate_and_save_cluster_config(self) -> bool: """Verify the config then write the values to prole.cfg.""" env_key = self._cluster_env_key() prev_service_ns = "" try: prev_service_ns = ( (self.prole_cfg_data.get("Global", {}) or {}) .get("SERVICE_NAMESPACE", "") .strip() ) except Exception: prev_service_ns = "" if env_key == "dev": cluster_val = self.selected_k3d_cluster.get() or "knoe-dev-cluster" if not cluster_val: messagebox.showwarning( "Validation", "Please select or create a k3d cluster." ) return False # Context switch (if context matches cluster name) try: subprocess.run( ["kubectl", "config", "use-context", f"k3d-{cluster_val}"], capture_output=True, text=True, ) except Exception: pass elif env_key == "service": # Service deployments can authenticate via an existing kubeconfig/context # (preferred), or via a token-based generated kubeconfig. kubeconfig = _find_kubeconfig_file() has_token_flow = bool((self.k3s_server_url.get() or "").strip()) and bool( (self.k3s_token.get() or "").strip() ) if not kubeconfig and not has_token_flow: messagebox.showwarning( "Validation", "Missing cluster credentials. Please select a working Kubernetes context (kubectx) " "or provide K3s server URL + token.", ) return False cluster_val = self.cluster_env.get() elif env_key == "prod": path = self.prod_artifacts_path.get().strip() if not path: messagebox.showwarning( "Validation", "Please specify an artifact staging directory." ) return False self._ensure_prod_config_state() put_result = self._prod_put_config(show_dialog=False) prod_errors = list(put_result.get("errors") or []) if prod_errors: messagebox.showwarning( "Validation", "Production config validation failed:\n\n" + "\n".join(prod_errors), ) return False cfg_vars = _collect_cfg_vars_from_data(getattr(self, "prole_cfg_data", None)) expanded_path = _expand_path_expr(path, cfg_vars) p = Path(expanded_path).expanduser() if not p.exists(): try: p.mkdir(parents=True, exist_ok=True) except Exception as e: messagebox.showerror( "Error", f"Failed to create directory {expanded_path or path}: {e}" ) return False cluster_val = self.cluster_env.get() else: cluster_val = self.cluster_env.get() # Capture cluster info for config. # ENVIRONMENT here is used as an env/mode hint; keep it as the selected env key. self.prole_cfg_data["Initialize Cluster"]["ENVIRONMENT"] = ( self.cluster_env.get() or "" ).strip() self.prole_cfg_data["Initialize Cluster"]["K3S_SERVER_URL"] = ( self.k3s_server_url.get() or "" ).strip() self.prole_cfg_data["Initialize Cluster"]["K3S_TOKEN"] = _encrypt_cfg_secret( self.k3s_token.get() or "" ) self.prole_cfg_data["Global"][ "SERVICE_NAMESPACE" ] = self._get_service_namespace() new_service_ns = self.prole_cfg_data["Global"]["SERVICE_NAMESPACE"] self.prole_cfg_data["Optional Features"]["SUPABASE_ENABLED"] = str( self.supabase_enabled.get() ) self.prole_cfg_data["Optional Features"]["GITOPS_ENABLED"] = str( self.gitops_enabled.get() ) self.prole_cfg_data["Optional Features"]["KERBEROS_ENABLED"] = str( self.kerberos_enabled.get() ) self.prole_cfg_data["Optional Features"]["AT_REST_ENCRYPTION_ENABLED"] = str( self.at_rest_encryption_enabled.get() ) if env_key == "prod": self.prole_cfg_data["Prod Cluster (k8s)"][ "ARTIFACTS_DIR" ] = self.prod_artifacts_path.get() try: prod_payload = self._prod_payload_from_vars() prod_plan = self.prod_config_api.post_plan(prod_payload) self.prole_cfg_data["Prod Cluster (k8s)"][ "PRODUCTION_CONFIG_YAML" ] = prod_plan.get("yaml", "") self.prole_cfg_data["Prod Cluster (k8s)"][ "OPENTOFU_VARS" ] = json.dumps( prod_plan.get("opentofuVars") or {}, sort_keys=True, separators=(",", ":") ) self.prole_cfg_data["Prod Cluster (k8s)"][ "INSTALL_PLAN" ] = "\n".join(prod_plan.get("installPlan") or []) self.prole_cfg_data["Prod Cluster (k8s)"][ "PLAN_MESSAGES" ] = self.prod_tab_messages.get() except Exception: pass # Save config self._save_prole_cfg() # Namespace change cleanup (prevents collisions when switching namespaces). try: prev = (prev_service_ns or "").strip() new = (new_service_ns or "").strip() if prev and new and prev != new: do_cleanup = False try: do_cleanup = messagebox.askyesno( "Namespace Changed", ( f"Common Core Services namespace changed from '{prev}' to '{new}'.\n\n" f"Reset the OLD namespace '{prev}' to avoid resource collisions?\n\n" "This deletes workloads/services in that namespace (PVCs are preserved)." ), ) except Exception: do_cleanup = False if do_cleanup: def _cleanup_worker(): try: self._set_cluster_env_message( f"Resetting old namespace: {prev}", "#ff9500", clear_after_ms=3000, ) try: self._ensure_k3s_kubeconfig_merged() except Exception: pass server, token = ("", "") try: server, token = self._k3s_connection_info() except Exception: server, token = ("", "") controller = getattr(self, "controller", None) project_root = getattr(controller, "project_root", None) if project_root: _reset_k3s_namespace(project_root, prev, server, token) self._set_cluster_env_message( f"Old namespace reset complete: {prev}", "#34c759", clear_after_ms=4000, ) except Exception as e: self._set_cluster_env_message( f"Namespace reset failed: {e}", "#ff3b30", clear_after_ms=6000, ) threading.Thread(target=_cleanup_worker, daemon=True).start() except Exception: pass # On save, immediately run a services status refresh so the UI reflects the # newly persisted namespace/cluster settings. try: self._verify_k3s_services() except Exception: pass return True def _cluster_status_snapshot(self) -> dict: env = self._cluster_env_key() cluster_ok = False if env == "dev": cluster_name = self.selected_k3d_cluster.get() or "knoe-dev-cluster" try: res = subprocess.run( ["k3d", "cluster", "list", "--no-headers"], capture_output=True, text=True, ) cluster_ok = cluster_name in (res.stdout or "") except Exception: cluster_ok = False cluster_msg = ( f"K3D Cluster ({cluster_name}): Running" if cluster_ok else f"K3D Cluster ({cluster_name}): Not found" ) else: cluster_ok, cluster_msg = self._check_k8s_cluster(env) cluster_fill = "#34c759" if cluster_ok else "#ff9f0a" common_ok = cluster_ok common_msg = ( "Common Services: Ready to deploy" if common_ok else "Common Services: Not ready" ) common_fill = "#34c759" if common_ok else "#ff9f0a" return { "env": env, "cluster_ok": cluster_ok, "cluster_msg": cluster_msg, "cluster_fill": cluster_fill, "common_ok": common_ok, "common_msg": common_msg, "common_fill": common_fill, } def _cluster_ready_for_navigation(self) -> bool: status = self._cluster_status_snapshot() if not status["cluster_ok"]: try: messagebox.showerror( "Cluster", "Cluster is not reachable. Please verify your cluster and try again.", ) except Exception: pass return False return True def check_cluster_status_async(self): def worker(): notice_active = False self._set_cluster_env_message("Collect Status", "#34c759") status = self._cluster_status_snapshot() self._set_cluster_env_message("Verify State", "#34c759") kong_status = None if status.get("cluster_ok"): kong_status = self._dashboard_kong_status(status.get("env")) self._set_cluster_env_message("Observe", "#34c759") # Observe-only: do not mutate the cluster automatically. anomalies = [] try: if kong_status and not kong_status.get("ok", True): anomalies.append("dashboard") except Exception: pass try: if self._authority_context_missing(): anomalies.append("authority") except Exception: pass if anomalies: self._set_cluster_env_message( "Notice: issues detected. Click Repair to reconcile.", "#ff9f0a", clear_after_ms=6000, ) notice_active = True def update_ui(): if hasattr(self, "k3d_status_label"): self.bg_canvas.itemconfig( self.k3d_status_label, text=status["cluster_msg"], fill=status["cluster_fill"], ) if hasattr(self, "common_services_status_label"): self.bg_canvas.itemconfig( self.common_services_status_label, text=status["common_msg"], fill=status["common_fill"], ) if hasattr(self, "_init_cluster_button"): try: self._init_cluster_button.configure(text="Save") except Exception: pass self.root.after(0, update_ui) # Clear message after a short delay if no notice is active if not notice_active: self._set_cluster_env_message("", "#34c759", clear_after_ms=2000) threading.Thread(target=worker, daemon=True).start() def _services_status_snapshot(self) -> dict: opentofu_ok = self._check_opentofu_health() opentofu_msg = "OpenTofu: Running" if opentofu_ok else "OpenTofu: Not reachable" opentofu_fill = "#34c759" if opentofu_ok else "#ff9f0a" return { "opentofu_msg": opentofu_msg, "opentofu_fill": opentofu_fill, } def check_services_status_async(self): def worker(): status = self._services_status_snapshot() def update_ui(): if hasattr(self, "_db_opentofu_status_label"): self.bg_canvas.itemconfig( self._db_opentofu_status_label, text=status["opentofu_msg"], fill=status["opentofu_fill"], ) self.root.after(0, update_ui) threading.Thread(target=worker, daemon=True).start() def ensure_cluster_ready(self): self._action_flags["init_cluster.start_cluster"] = True # Implementation of cluster creation/startup def worker(): cluster_env = self._cluster_env_key() if cluster_env == "dev": # 1. Start Docker if not running if not self.controller.check_docker_running(): # Attempt to start Docker on macOS subprocess.run(["open", "-a", "Docker"], capture_output=True) # Wait for it to start for _ in range(30): time.sleep(2) if self.controller.check_docker_running(): break if not self.controller.check_docker_running(): self.root.after( 0, lambda: messagebox.showerror( "Docker", "Could not start Docker. Please start it manually.", ), ) return # 2. Manage or verify cluster cluster_name = "knoe-dev-cluster" res = subprocess.run( ["k3d", "cluster", "list", "--no-headers"], capture_output=True, text=True, ) res_stdout = res.stdout or "" if self._reset_cluster: subprocess.run( ["k3d", "cluster", "delete", cluster_name], capture_output=True ) res_stdout = "" self._reset_cluster = False if cluster_name not in res_stdout: # Create it # Default args based on README.md prole_data = str(self._resolve_env_dir("PROLE_DATA", "data")) volume_args = _k3d_prole_data_volume_args(prole_data) cmd = [ "k3d", "cluster", "create", cluster_name, "-a", "2", ] + volume_args reg_args = [] try: if self.ensure_local_registry_available(): reg_args = ["--registry-use", "k3d-prole-registry:5000"] except Exception: reg_args = [] cmd += reg_args + ["--api-port", "0.0.0.0:6443"] # Run in terminal or capture output? Let's use a console window later. # For now, run it and update status. subprocess.run(cmd, capture_output=True) else: # Start it if it's stopped subprocess.run( ["k3d", "cluster", "start", cluster_name], capture_output=True ) self.check_cluster_status_async() self.root.after( 0, lambda: messagebox.showinfo( "Cluster", f"Cluster {cluster_name} is ready." ), ) else: if self._reset_cluster and cluster_env in ("service", "k3s"): ns = (self._get_service_namespace() or "").strip() or "default" server, token = self._k3s_connection_info() _reset_k3s_namespace( self.controller.project_root, ns, server, token ) self._reset_cluster = False ok, msg = self._check_k8s_cluster(cluster_env) if not ok: self.root.after(0, lambda: messagebox.showerror("Cluster", msg)) return self.check_cluster_status_async() self.root.after(0, lambda: messagebox.showinfo("Cluster", msg)) threading.Thread(target=worker, daemon=True).start() def _verify_k3s_services(self): if not self._k3s_service_status_console: self._stop_k3s_service_status_updates() return try: self._k3s_service_status_console.clear() except Exception: pass try: if self._k3s_service_notebook and self._k3s_service_status_tab: self._k3s_service_notebook.select(self._k3s_service_status_tab) except Exception: pass self._start_k3s_service_status_updates() def _start_k3s_service_status_updates(self): if not self._k3s_status_active: self._k3s_status_active = True if self._k3s_status_after_id: try: self.root.after_cancel(self._k3s_status_after_id) except Exception: pass self._k3s_status_after_id = None self._schedule_k3s_service_status_update(0) def _stop_k3s_service_status_updates(self): self._k3s_status_active = False self._k3s_status_inflight = False if self._k3s_status_after_id: try: self.root.after_cancel(self._k3s_status_after_id) except Exception: pass self._k3s_status_after_id = None def _schedule_k3s_service_status_update(self, delay_ms: int): if not self._k3s_status_active: return if not self.root or not self.root.winfo_exists(): return try: self._k3s_status_after_id = self.root.after( delay_ms, self._run_k3s_service_status_once ) except Exception: self._k3s_status_after_id = None def _run_k3s_service_status_once(self): if self._k3s_status_inflight: return if not self._k3s_service_status_console: self._k3s_status_inflight = False self.safe_after(lambda: self._schedule_k3s_service_status_update(30000)) return self._k3s_status_inflight = True console = self._k3s_service_status_console console.clear() console.write( f"Service status refresh @ {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n" ) def worker(): collected_lines = [] try: ns = self._get_service_namespace() env = self._script_env_for_namespace(ns) env["PROLE_MODE"] = self._deployment_mode() if not env.get("KUBECONFIG"): console.write( "KUBECONFIG not generated. Check cluster credentials in prole.cfg.\n" ) return verify_args = ["-n", ns, "verify"] def _on_line(line): console.write(line) collected_lines.append(line) rc = self.controller.run_script( "init_common_services.sh", args=verify_args, env=env, on_line=_on_line, ) console.write(f"\nVerification Exit status: {rc}\n") # Parse per-component traffic light status from [OK]/[FAIL] lines component_status = self._parse_service_status_lines(collected_lines) all_green = rc == 0 def _update_traffic_lights(): self._all_services_green = all_green # Ensure common services success flag is synced for navigation self._common_services_success = all_green lights = getattr(self, "_service_traffic_lights", {}) for comp_key, indicator_id in lights.items(): if comp_key in component_status: color = ( "#34c759" if component_status[comp_key] else "#ff3b30" ) else: color = "#34c759" if all_green else "#8e8e93" try: self.bg_canvas.itemconfig( indicator_id, fill=color, outline=color ) except Exception: pass # Update footer to enable Next button if everything is green self.update_footer() self.safe_after(_update_traffic_lights) finally: self._k3s_status_inflight = False self.safe_after(lambda: self._schedule_k3s_service_status_update(30000)) threading.Thread(target=worker, daemon=True).start() def _parse_service_status_lines(self, lines: list[str]) -> dict[str, bool]: """Parse [OK]/[FAIL] lines from status_common_services.sh output. Returns a dict mapping component keys (registry, certmgr, garage, kong, openbao, opentofu) to True (healthy) or False (unhealthy). """ # Map resource names to traffic light component keys name_map = { "registry": "registry", "opentofu": "opentofu", "garage": "garage", "openbao": "openbao", "prole-svc-kong": "kong", "cert-manager": "certmgr", "cert-manager-cainjector": "certmgr", "cert-manager-webhook": "certmgr", } # Start with all components healthy (True); any FAIL sets to False result: dict[str, bool] = {} import re as _re for line in lines: m = _re.match(r"\[(OK|FAIL)\]\s+(\S+)/(\S+)", line) if not m: continue status_tag = m.group(1) resource_name = m.group(3) comp_key = name_map.get(resource_name) if not comp_key: continue if comp_key not in result: result[comp_key] = True if status_tag == "FAIL": result[comp_key] = False return result def _deploy_k3s_services(self): """Deploy common services (Registry/OpenTofu/Garage) to remote k3s cluster.""" def _deploy(): console = self._k3s_service_deploy_console if console: console.clear() console.write( f"Deploying common services @ {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n" ) try: if self._k3s_service_notebook and self._k3s_service_deploy_tab: self._k3s_service_notebook.select(self._k3s_service_deploy_tab) except Exception: pass service_ns = self._get_service_namespace() env = self._script_env_for_namespace(service_ns) env["PROLE_MODE"] = self._deployment_mode() log_path = self._common_services_log_path() env["COMMON_SERVICES_INIT_LOG"] = str(log_path) try: log_fp = log_path.open("a", encoding="utf-8") log_fp.write( f"\n# Common services deploy @ {time.strftime('%Y-%m-%d %H:%M:%S')}\n" ) log_fp.flush() except Exception: log_fp = None if console: console.write(f"Log file: {log_path}\n\n") if not env.get("KUBECONFIG"): if console: console.write( "KUBECONFIG not generated. Check cluster credentials in prole.cfg.\n" ) if log_fp: try: log_fp.write( "KUBECONFIG not generated. Check cluster credentials in prole.cfg.\n" ) log_fp.close() except Exception: pass self._verify_k3s_services() return try: common_args = ["-n", env["NAMESPACE"]] if self.kerberos_enabled.get(): common_args.append("-k") common_args.append("update") def _log_line(line: str): if console: console.write(line) if log_fp: try: log_fp.write(line) log_fp.flush() except Exception: pass rc = self.controller.run_script( "init_common_services.sh", args=common_args, env=env, on_line=_log_line, ) if console: console.write(f"\nExit status: {rc}\n") if log_fp: try: log_fp.write(f"\nExit status: {rc}\n") except Exception: pass finally: if log_fp: try: log_fp.close() except Exception: pass self._verify_k3s_services() threading.Thread(target=_deploy, daemon=True).start() def _ensure_k3s_kubeconfig_merged(self): """Ensure a valid kubeconfig is available and merged into ~/.kube/config. Prefers an existing Ansible-fetched kubeconfig (with client-certificate auth) over generating a new token-based one. """ # 1. Check for existing cert-based kubeconfig first existing = _find_kubeconfig_file() if existing: self._managed_kubeconfig = existing _merge_kubeconfig(existing) return # 2. Fall back to token-based generation server, token = self._k3s_connection_info() if not server or not token: return kubeconfig_path = str(_write_k3s_kubeconfig(server, token)) self._managed_kubeconfig = kubeconfig_path _merge_kubeconfig(kubeconfig_path) def _k3s_connection_info(self) -> tuple[str, str]: # Check UI fields first; fall back to the shared 3-source resolver. ui_server = (_safe_str(self.k3s_server_url.get()) or "").strip() ui_token = (_safe_str(self.k3s_token.get()) or "").strip() if ui_token: ui_token = self._resolve_secret_value(ui_token) ui_token = _normalize_k3s_token(ui_token) if ui_server and not ui_server.startswith("http"): ui_server = f"https://{ui_server}" # If UI fields are populated, use them directly. if ui_server and ui_token: return ui_server, ui_token # Otherwise delegate to the unified resolver (env → cfg → ansible). resolved_server, resolved_token = _resolve_k3s_connection_fn( project_root=PROJECT_ROOT, ) return ui_server or resolved_server, ui_token or resolved_token def _k3s_connection_raw(self) -> tuple[str, str]: """Return (server_url, token) without attempting secret resolution.""" server = ( _safe_str(self.k3s_server_url.get()) or os.environ.get("PROLE_K3S_SERVER") or os.environ.get("K3S_SERVER_URL") or "" ).strip() token = ( _safe_str(self.k3s_token.get()) or os.environ.get("PROLE_K3S_TOKEN") or os.environ.get("K3S_TOKEN") or "" ).strip() if server and not server.startswith("http"): server = f"https://{server}" return server, token