""" Managed Kubernetes cluster operations. """ import json import subprocess import hashlib import getpass from pathlib import Path from typing import Optional, List, Dict, Literal from .utils import Colors, Logger from .config import KubeManagerConfig KubeconfigMode = Literal["minikube", "current"] class KubeManager: """Manages Kubernetes cluster lifecycle and TiDB Operator installation.""" OPERATOR_VERSION = 'v2.1.0-alpha.7' def __init__(self, config: KubeManagerConfig): """ Initialize KubeManager. Args: config: KubeManagerConfig instance """ self.tiop_home = config.tiop_home self.binary_manager = config.binary_manager self.minikube_config = config.minikube_config self.kubeconfig_state_file = config.kubeconfig_state_file self.run_kubectl = config.run_kubectl # Use base kubectl runner (no status check) self.run_minikube = config.run_minikube # Generate unique minikube profile name self.minikube_profile = self._get_or_create_profile_name() def _get_or_create_profile_name(self) -> str: """Generate or retrieve a unique minikube profile name for this tiop installation.""" # First check if profile is saved in kubeconfig state state = self._load_kubeconfig_state() if state and state.get("minikube_profile"): return state["minikube_profile"] # Fallback to old profile file for backward compatibility profile_file = self.tiop_home / '.minikube-profile' if profile_file.exists(): profile_name = profile_file.read_text().strip() # Migrate to new state file self._update_kubeconfig_state(minikube_profile=profile_name) # Remove old file after migration profile_file.unlink() Logger.info(f"Migrated profile name to kubeconfig state: {profile_name}") return profile_name # Generate unique profile based on username and tiop_home path username = getpass.getuser() home_hash = hashlib.sha256(str(self.tiop_home).encode()).hexdigest()[:8] profile_name = f"tiop-{username}-{home_hash}" # Save to state file self._update_kubeconfig_state(minikube_profile=profile_name) Logger.info(f"Generated unique minikube profile: {profile_name}") return profile_name def _save_kubeconfig_state(self, mode: KubeconfigMode, source: Optional[str] = None, profile: Optional[str] = None): """Save the current kubeconfig state to a structured JSON file.""" self._update_kubeconfig_state(source=source,mode=mode) def _load_kubeconfig_state(self) -> Optional[Dict]: """Load the kubeconfig state from the JSON file.""" if not self.kubeconfig_state_file.exists(): return None try: return json.loads(self.kubeconfig_state_file.read_text()) except (json.JSONDecodeError, Exception) as e: Logger.warn(f"Failed to load kubeconfig state: {e}") return None def get_kubeconfig_mode(self) -> Optional[KubeconfigMode]: """Get the current kubeconfig mode.""" state = self._load_kubeconfig_state() return state.get("mode") if state else None def get_effective_kubeconfig_mode(self) -> KubeconfigMode: """Get the effective kubeconfig mode, defaulting to 'minikube' if not set.""" mode = self.get_kubeconfig_mode() if mode is None: # Default to minikube for safety - we don't want to accidentally affect user's cluster return "minikube" return mode def get_kubeconfig_source(self) -> Optional[str]: """Get the source of the current kubeconfig (e.g., file path for custom mode).""" state = self._load_kubeconfig_state() return state.get("source") if state else None def get_minikube_profile(self) -> Optional[str]: """Get the minikube profile name from state file.""" state = self._load_kubeconfig_state() return state.get("minikube_profile") if state else None def _update_kubeconfig_state(self, **kwargs): """Update specific fields in the kubeconfig state without overwriting others.""" state = self._load_kubeconfig_state() or {} state.update(kwargs) self.kubeconfig_state_file.write_text(json.dumps(state, indent=2)) def _is_minikube_running(self) -> bool: """Check if the minikube cluster is running.""" try: result = self.run_minikube(['status', '-p', self.minikube_profile], capture_output=True, check=False) # Check if the output contains "Running" status output = result.stdout.lower() if result.stdout else "" return result.returncode == 0 and "running" in output except Exception: return False def _ensure_minikube_running(self): """Ensure minikube cluster is running, or raise an error.""" if not self._is_minikube_running(): Logger.die( f"Minikube cluster (profile: {self.minikube_profile}) is not running.\n" f"Please start it with: tiop managed-kube init\n" f"Or check status with: tiop managed-kube status" ) def is_using_custom_kubeconfig(self) -> bool: """Check if currently using a custom kubeconfig.""" mode = self.get_kubeconfig_mode() # Check both new structured config and old marker file for backward compatibility return mode == "custom" or self.custom_config_marker.exists() def is_using_current_context(self) -> bool: """Check if currently using the current kubectl context (no kubeconfig injection).""" return self.get_kubeconfig_mode() == "current" def use(self, use_minikube: bool = False, use_current: bool = False): """Switch between minikube and current context.""" if use_minikube: # Switch back to minikube current_mode = self.get_kubeconfig_mode() if current_mode == "minikube": Logger.info("Already using minikube kubeconfig") return Logger.info("Switching back to minikube kubeconfig...") # Save state self._save_kubeconfig_state("minikube") Logger.success("Switched to minikube kubeconfig") Logger.info(f"Using: {self.minikube_config}") elif use_current: # Switch to use current kubectl context (no kubeconfig injection) current_mode = self.get_kubeconfig_mode() if current_mode == "current": Logger.info("Already using current kubectl context") return Logger.info("Switching to use current kubectl context...") # Save state self._save_kubeconfig_state("current") Logger.success("Switched to use current kubectl context") Logger.info("TiOP will now use your current kubectl context without injecting KUBECONFIG") Logger.warn("Note: 'managed-kube init', 'status', and 'destroy' are disabled when using current context") Logger.info("To switch back: tiop managed-kube use --minikube") else: # No flags provided, default to minikube for safety Logger.info("No flags provided, defaulting to minikube mode...") self.use(use_minikube=True) def init(self, extra_args: Optional[List[str]] = None): """Initialize managed Kubernetes cluster.""" mode = self.get_effective_kubeconfig_mode() # Set default mode to minikube if not already set if self.get_kubeconfig_mode() is None: self._save_kubeconfig_state("minikube") Logger.info("Defaulting to minikube mode") if mode == "current": Logger.die(f"Cannot use 'init' with current context mode. Switch back with: tiop managed-kube use --minikube") Logger.info("Initializing managed Kubernetes cluster...") Logger.info(f"Using minikube profile: {self.minikube_profile}") # Ensure all binaries are available self.binary_manager.ensure_all_binaries() # Check if minikube cluster exists try: result = self.run_minikube(['status', '-p', self.minikube_profile], capture_output=True, check=False) if result.returncode == 0: Logger.info("Minikube cluster already exists") else: raise Exception("Cluster not running") except Exception: Logger.info("Creating new minikube cluster...") minikube_args = ['start', '-p', self.minikube_profile, '--kubernetes-version=stable'] if extra_args: minikube_args.extend(extra_args[1:]) Logger.info(f"Additional minikube arguments: {' '.join(extra_args[1:])}") self.run_minikube(minikube_args) # Export kubeconfig using minikube's update-context to ensure we get the right context # IMPORTANT: Set KUBECONFIG to our minikube config path to avoid overwriting user's current kubeconfig Logger.info(f"Updating kubeconfig context for minikube profile: {self.minikube_profile}") self.run_minikube(['update-context', '-p', self.minikube_profile]) # Now export the exact minikube kubeconfig for this profile Logger.info(f"Exporting kubeconfig to {self.minikube_config}") result = self.run_minikube(['kubectl', '-p', self.minikube_profile, '--', 'config', 'view', '--flatten', '--minify'], capture_output=True) self.minikube_config.write_text(result.stdout) # Wait for cluster to be ready Logger.info("Waiting for cluster to be ready...") self.run_kubectl([ 'wait', '--for=condition=Ready', 'nodes', '--all', '--timeout=300s' ]) Logger.success("Kubernetes cluster is ready") # Install operator self.install_operator() def install_operator(self): """Install TiDB Operator.""" Logger.info(f"Installing TiDB Operator {self.OPERATOR_VERSION}...") # Apply CRDs Logger.info("Applying TiDB Operator CRDs...") crd_url = f"https://github.com/pingcap/tidb-operator/releases/download/{self.OPERATOR_VERSION}/tidb-operator.crds.yaml" self.run_kubectl(['apply', '-f', crd_url, '--server-side']) # Apply operator Logger.info("Applying TiDB Operator...") operator_url = f"https://github.com/pingcap/tidb-operator/releases/download/{self.OPERATOR_VERSION}/tidb-operator.yaml" self.run_kubectl(['apply', '-f', operator_url, '--server-side']) # Wait for operator to be ready Logger.info("Waiting for TiDB Operator to be ready...") try: self.run_kubectl([ 'wait', '--for=condition=Ready', 'pods', '-n', 'tidb-admin', '-l', 'app.kubernetes.io/component=controller-manager', '--timeout=300s' ], check=False) except Exception: Logger.warn("Operator may not be fully ready yet") Logger.success(f"TiDB Operator {self.OPERATOR_VERSION} installed successfully") def status(self): """Show Kubernetes cluster status.""" mode = self.get_effective_kubeconfig_mode() if mode == "current": Logger.die(f"Cannot use 'status' with current context mode. Switch back with: tiop managed-kube use --minikube") Logger.info("Checking Kubernetes cluster status...") Logger.info(f"Using minikube profile: {self.minikube_profile}") try: result = self.run_minikube(['status', '-p', self.minikube_profile], check=False) if result.returncode != 0: Logger.error("Minikube cluster is not running") return except Exception as e: Logger.error(f"Failed to check minikube status: {e}") return Logger.info("\nChecking TiDB Operator status...") try: # Check if tidb-admin namespace exists result = self.run_kubectl(['get', 'namespace', 'tidb-admin'], capture_output=True, check=False) if result.returncode == 0: self.run_kubectl(['get', 'pods', '-n', 'tidb-admin']) else: Logger.warn("TiDB Operator is not installed") except Exception as e: Logger.error(f"Failed to check operator status: {e}") def destroy(self): """Destroy minikube cluster.""" mode = self.get_effective_kubeconfig_mode() if mode == "current": Logger.die(f"Cannot use 'destroy' with current context mode. Switch back with: tiop managed-kube use --minikube") Logger.info(f"This will destroy the minikube cluster (profile: {self.minikube_profile})") response = input( f"{Colors.YELLOW}and clean kubeconfig and local cluster configs under {self.tiop_home}. Are you sure? (y/N){Colors.NC} " ) if response.lower() in ['y', 'yes']: Logger.info(f"Destroying minikube cluster with profile: {self.minikube_profile}...") self.run_minikube(['delete', '-p', self.minikube_profile]) Logger.success("Minikube cluster destroyed") # Cleanup kubeconfig and local cluster configs under tiop_home try: removed_any = False # Remove kubeconfig file if present if self.minikube_config.exists(): try: self.minikube_config.unlink() Logger.info(f"Removed kubeconfig: {self.minikube_config}") removed_any = True except Exception as e: Logger.warn(f"Failed to remove kubeconfig {self.minikube_config}: {e}") # Remove generated cluster manifests (*.yaml) and config refs (*.config) for pattern in ("*.yaml", "*.config"): for f in sorted(self.tiop_home.glob(pattern)): # Be careful not to remove anything inside bin/ or other directories if f.is_file(): # Skip files that are clearly not cluster artifacts if f.name == self.minikube_config.name: continue try: f.unlink() Logger.info(f"Removed local cluster file: {f}") removed_any = True except Exception as e: Logger.warn(f"Failed to remove file {f}: {e}") # Clean up the kubeconfig state file if self.kubeconfig_state_file.exists(): try: self.kubeconfig_state_file.unlink() Logger.info(f"Removed kubeconfig state: {self.kubeconfig_state_file}") removed_any = True except Exception as e: Logger.warn(f"Failed to remove kubeconfig state {self.kubeconfig_state_file}: {e}") if removed_any: Logger.success("Cleaned kubeconfig and local cluster configs under tiop_home") else: Logger.info("No kubeconfig or local cluster configs found to clean") except Exception as e: Logger.warn(f"Cleanup encountered an issue: {e}") else: Logger.info("Operation cancelled") def download_deps(self, force: bool = False): """Download all required dependencies.""" Logger.info("Downloading required dependencies...") binaries = ['helm', 'minikube', 'kubectl', 'jsonnet'] for binary in binaries: if force: Logger.info(f"Force downloading {binary}...") self.binary_manager.download_binary(binary, force=True) else: local_path = self.binary_manager.get_binary_path(binary) if local_path.exists() and local_path.stat().st_mode & 0o111: Logger.info(f"{binary} is already downloaded in {self.tiop_home / 'bin'}") else: self.binary_manager.download_binary(binary) Logger.success("All dependencies are ready") Logger.info(f"Binaries are located in: {self.tiop_home / 'bin'}") # Show versions Logger.info("\nInstalled versions:") for binary in binaries: version = self.binary_manager.VERSIONS.get(binary, 'unknown') local_path = self.binary_manager.get_binary_path(binary) status = "✓" if (local_path.exists() and local_path.stat().st_mode & 0o111) else "✗" Logger.info(f" {status} {binary}: {version}")