#!/usr/bin/env python3 """ TiOP - TiDB Operator Tool A Python-based tool to help deploy TiDB clusters in Kubernetes. Automatically downloads and manages required binaries (helm, minikube, kubectl, jsonnet). Credit: Claude Sonnet 4.5 """ import os import sys import argparse from pathlib import Path from typing import Optional, Dict from tiop_core import ( Logger, LogLevel, BinaryManager, ClusterManager, KubeManager, ClusterManagerConfig, KubeManagerConfig ) class TiOP: """Main TiOP tool class.""" def __init__(self, namespace: Optional[str] = None): # Support both TiOP-specific env var and a common misspelling/alias 'tiup_home' self.tiop_home = Path(os.environ.get('tiup_home', Path.home() / '.tiop')) self.tiop_home.mkdir(parents=True, exist_ok=True) self.minikube_config = self.tiop_home / 'minikubeconfig.yml' self.kubeconfig_state_file = self.tiop_home / 'kubeconfig-state.json' self.script_dir = Path(os.path.realpath(__file__)).parent self.binary_manager = BinaryManager(self.tiop_home) # Create base runners (without checks or profile injection) def base_run_minikube(args, **kwargs): return self.binary_manager.run('minikube', args, env=self.get_kubeconfig_env(), **kwargs) def base_run_kubectl(args, **kwargs): """Base kubectl runner without minikube status check.""" return self.binary_manager.run('kubectl', args, env=self.get_kubeconfig_env(), **kwargs) # Initialize KubeManager with base runners kube_config = KubeManagerConfig( tiop_home=self.tiop_home, binary_manager=self.binary_manager, minikube_config=self.minikube_config, kubeconfig_state_file=self.kubeconfig_state_file, run_kubectl=base_run_kubectl, # Use base runner for KubeManager run_minikube=base_run_minikube # Use base runner for KubeManager ) self.kube_manager = KubeManager(kube_config) cluster_config = ClusterManagerConfig( tiop_home=self.tiop_home, script_dir=self.script_dir, binary_manager=self.binary_manager, run_kubectl=self.run_kubectl, run_jsonnet=self.run_jsonnet, namespace=namespace ) self.cluster_manager = ClusterManager(cluster_config) Logger.trace(f"TiOP home: {self.tiop_home}") # Show current kubeconfig mode and profile mode = self.kube_manager.get_effective_kubeconfig_mode() if mode == "current": Logger.info(f"Using current kubectl context (no kubeconfig injection)") elif mode == "minikube": Logger.info(f"Using minikube kubeconfig (profile: {self.kube_manager.minikube_profile})") if namespace is not None: Logger.info(f"Using namespace: {namespace}") def _resolve_config_path(self, config: Optional[str], playground: bool, x_playground: bool) -> Optional[str]: """Resolve config path from flags.""" if playground or x_playground: # Playground configs are only allowed with minikube backend mode = self.kube_manager.get_effective_kubeconfig_mode() if mode != "minikube": Logger.die(f"Playground configurations are only available when using the minikube backend.\n" f"Current backend: {mode}\n" f"Switch to minikube with: tiop.py managed-kube use --minikube") if playground: playground_config = self.script_dir / 'playground' / 'normal.jsonnet' if not playground_config.exists(): Logger.die(f"Playground config not found: {playground_config}") return str(playground_config) elif x_playground: playground_config = self.script_dir / 'playground' / 'next-gen.jsonnet' if not playground_config.exists(): Logger.die(f"Next-gen playground config not found: {playground_config}") return str(playground_config) else: return config def get_kubeconfig_env(self) -> Dict[str, str]: """Get environment with KUBECONFIG set.""" mode = self.kube_manager.get_effective_kubeconfig_mode() # If using current context, don't inject KUBECONFIG if mode == "current": return {} # Otherwise use minikube config if self.minikube_config.exists(): return {'KUBECONFIG': str(self.minikube_config)} return {} def run_kubectl(self, args, **kwargs): """Run kubectl with proper kubeconfig.""" # Check if minikube is running when using minikube mode mode = self.kube_manager.get_effective_kubeconfig_mode() if mode == "minikube": self.kube_manager._ensure_minikube_running() return self.binary_manager.run('kubectl', args, env=self.get_kubeconfig_env(), **kwargs) def run_helm(self, args, **kwargs): """Run helm with proper kubeconfig.""" return self.binary_manager.run('helm', args, env=self.get_kubeconfig_env(), **kwargs) def run_minikube(self, args, **kwargs): """Run minikube with automatic profile injection.""" # Auto-inject profile if not already specified and kube_manager is available if hasattr(self, 'kube_manager') and '-p' not in args and '--profile' not in args: # Commands that should use the profile profile_commands = ['status', 'start', 'stop', 'delete', 'pause', 'unpause', 'kubectl', 'ssh', 'service', 'tunnel', 'logs', 'ip', 'dashboard', 'addons', 'config'] if args and args[0] in profile_commands: args = [args[0], '-p', self.kube_manager.minikube_profile] + args[1:] return self.binary_manager.run('minikube', args, **kwargs) def run_jsonnet(self, args, **kwargs): """Run jsonnet.""" return self.binary_manager.run('jsonnet', args, **kwargs) # Direct tool wrappers def kubectl(self, args): """Run kubectl directly.""" self.run_kubectl(args) def helm(self, args): """Run helm directly.""" self.run_helm(args) def minikube(self, args): """Run minikube directly.""" self.run_minikube(args) def main(): """Main entry point.""" parser = argparse.ArgumentParser( description='TiOP - TiDB Operator Tool', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Download all required dependencies tiop.py managed-kube download-deps # Force re-download dependencies tiop.py managed-kube download-deps --force # Initialize Kubernetes cluster with TiDB Operator tiop.py managed-kube init # Initialize with custom minikube settings tiop.py managed-kube init -- --cpus=4 --memory=8192 --driver=docker # Use current kubectl context (no kubeconfig injection) tiop.py managed-kube use --current # Switch to minikube (default if no flags provided) tiop.py managed-kube use # or explicitly: tiop.py managed-kube use --minikube # Inspect configuration before creating tiop.py cluster inspect my-cluster --config=./my-config.jsonnet # Create a TiDB cluster with custom config tiop.py cluster create my-cluster --config=./my-config.jsonnet # Create a cluster in a specific namespace tiop.py cluster create my-cluster --config=./my-config.jsonnet -n my-namespace # Create a cluster with normal playground config tiop.py cluster create my-cluster --playground # Create a cluster with next-gen playground config tiop.py cluster create my-cluster --x-playground # Check cluster status tiop.py cluster status my-cluster # List all clusters tiop.py cluster list # Update cluster (uses saved config) tiop.py cluster apply my-cluster # Delete a cluster (uses saved YAML) tiop.py cluster delete my-cluster # List available templates tiop.py cluster template # View a specific template tiop.py cluster template normal # Use kubectl through tiop tiop.py kubectl get pods # Set log level to trace for verbose output tiop.py --log-level trace cluster status my-cluster # Set log level to error for minimal output tiop.py --log-level error cluster list """ ) # Add global namespace argument parser.add_argument('-n', '--namespace', default=None, help='Kubernetes namespace (if not specified, uses current context namespace)') # Add global log level argument parser.add_argument('--log-level', '-l', choices=['trace', 'info', 'success', 'warn', 'error'], default='info', help='Set log level (default: info)') subparsers = parser.add_subparsers(dest='command', help='Command to execute') # managed-kube commands (with 'kube' alias) managed_kube = subparsers.add_parser('managed-kube', aliases=['kube'], help='Manage Kubernetes cluster') managed_kube_subs = managed_kube.add_subparsers(dest='subcommand') use_parser = managed_kube_subs.add_parser('use', help='Switch between minikube and current context (defaults to minikube)') use_group = use_parser.add_mutually_exclusive_group(required=False) use_group.add_argument('--minikube', action='store_true', help='Switch to minikube kubeconfig (default)') use_group.add_argument('--current', action='store_true', help='Use current kubectl context (no kubeconfig injection)') init_parser = managed_kube_subs.add_parser('init', help='Initialize minikube cluster') init_parser.add_argument('init_args', nargs=argparse.REMAINDER, help='Additional arguments to pass to minikube start') managed_kube_subs.add_parser('install-operator', help='Install TiDB Operator') managed_kube_subs.add_parser('status', help='Show cluster status') managed_kube_subs.add_parser('destroy', help='Destroy minikube cluster') download_deps = managed_kube_subs.add_parser('download-deps', help='Download all required dependencies') download_deps.add_argument('--force', action='store_true', help='Force re-download even if binaries exist') # cluster commands cluster = subparsers.add_parser('cluster', help='Manage TiDB clusters') cluster_subs = cluster.add_subparsers(dest='subcommand') cluster_create = cluster_subs.add_parser('create', help='Create a TiDB cluster') cluster_create.add_argument('name', help='Cluster name') config_group_create = cluster_create.add_mutually_exclusive_group() config_group_create.add_argument('--config', '-f', help='Config file path') config_group_create.add_argument('--playground', action='store_true', help='Use playground/normal.jsonnet config') config_group_create.add_argument('--x-playground', action='store_true', help='Use playground/next-gen.jsonnet config') cluster_apply = cluster_subs.add_parser('apply', help='Apply or update a TiDB cluster') cluster_apply.add_argument('name', help='Cluster name') config_group_apply = cluster_apply.add_mutually_exclusive_group() config_group_apply.add_argument('--config', '-f', help='Config file path') config_group_apply.add_argument('--playground', action='store_true', help='Use playground/normal.jsonnet config') config_group_apply.add_argument('--x-playground', action='store_true', help='Use playground/next-gen.jsonnet config') cluster_delete = cluster_subs.add_parser('delete', help='Delete a TiDB cluster') cluster_delete.add_argument('name', help='Cluster name') cluster_status = cluster_subs.add_parser('status', help='Show cluster status') cluster_status.add_argument('name', help='Cluster name') cluster_subs.add_parser('list', help='List all TiDB clusters') cluster_inspect = cluster_subs.add_parser('inspect', help='Inspect cluster configuration without applying') cluster_inspect.add_argument('name', nargs='?', help='Cluster name (optional if --config provided)') config_group_inspect = cluster_inspect.add_mutually_exclusive_group() config_group_inspect.add_argument('--config', '-f', help='Config file path') config_group_inspect.add_argument('--playground', action='store_true', help='Use playground/normal.jsonnet config') config_group_inspect.add_argument('--x-playground', action='store_true', help='Use playground/next-gen.jsonnet config') cluster_template = cluster_subs.add_parser('template', help='List or display configuration templates') cluster_template.add_argument('template_name', nargs='?', help='Template name (omit to list all templates)') # Direct tool wrappers kubectl_parser = subparsers.add_parser('kubectl', help='Run kubectl with TIOP config') kubectl_parser.add_argument('kubectl_args', nargs=argparse.REMAINDER) helm_parser = subparsers.add_parser('helm', help='Run helm with TIOP config') helm_parser.add_argument('helm_args', nargs=argparse.REMAINDER) minikube_parser = subparsers.add_parser('minikube', help='Run minikube') minikube_parser.add_argument('minikube_args', nargs=argparse.REMAINDER) args = parser.parse_args() if not args.command: parser.print_help() return # Set log level based on argument log_level_str = getattr(args, 'log_level', 'info').upper() log_level_map = { 'TRACE': LogLevel.TRACE, 'INFO': LogLevel.INFO, 'SUCCESS': LogLevel.SUCCESS, 'WARN': LogLevel.WARN, 'ERROR': LogLevel.ERROR, } Logger.set_level(log_level_map.get(log_level_str, LogLevel.INFO)) namespace = getattr(args, 'namespace', None) tiop = TiOP(namespace=namespace) try: if args.command in ('managed-kube', 'kube'): if args.subcommand == 'use': use_minikube = getattr(args, 'minikube', False) use_current = getattr(args, 'current', False) tiop.kube_manager.use(use_minikube=use_minikube, use_current=use_current) elif args.subcommand == 'init': extra_args = getattr(args, 'init_args', None) tiop.kube_manager.init(extra_args) elif args.subcommand == 'install-operator': tiop.kube_manager.install_operator() elif args.subcommand == 'status': tiop.kube_manager.status() elif args.subcommand == 'destroy': tiop.kube_manager.destroy() elif args.subcommand == 'download-deps': force = getattr(args, 'force', False) tiop.kube_manager.download_deps(force=force) else: Logger.error("Unknown managed-kube subcommand") sys.exit(1) elif args.command == 'cluster': if args.subcommand == 'create': playground = getattr(args, 'playground', False) x_playground = getattr(args, 'x_playground', False) config = tiop._resolve_config_path(args.config, playground, x_playground) tiop.cluster_manager.create(args.name, config) elif args.subcommand == 'apply': playground = getattr(args, 'playground', False) x_playground = getattr(args, 'x_playground', False) config = tiop._resolve_config_path(args.config, playground, x_playground) tiop.cluster_manager.apply(args.name, config) elif args.subcommand == 'delete': tiop.cluster_manager.delete(args.name) elif args.subcommand == 'status': tiop.cluster_manager.status(args.name) elif args.subcommand == 'list': tiop.cluster_manager.list() elif args.subcommand == 'inspect': name = getattr(args, 'name', None) playground = getattr(args, 'playground', False) x_playground = getattr(args, 'x_playground', False) config = tiop._resolve_config_path(getattr(args, 'config', None), playground, x_playground) tiop.cluster_manager.inspect(name, config) elif args.subcommand == 'template': template_name = getattr(args, 'template_name', None) tiop.cluster_manager.template(template_name) else: Logger.error("Unknown cluster subcommand") sys.exit(1) elif args.command == 'kubectl': tiop.kubectl(args.kubectl_args) elif args.command == 'helm': tiop.helm(args.helm_args) elif args.command == 'minikube': tiop.minikube(args.minikube_args) else: Logger.error(f"Unknown command: {args.command}") parser.print_help() sys.exit(1) except KeyboardInterrupt: Logger.info("\nOperation cancelled by user") sys.exit(130) except Exception as e: Logger.die(f"Unexpected error: {e}") if __name__ == '__main__': main()