diff --git a/README.md b/README.md index 6fe9c04..ee58778 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ ## Особенности - 🎯 Интерактивное меню с навигацией стрелками -- 🔄 Переключение между Kubernetes контекстами/кластерами +- 🔄 Полное управление Kubernetes контекстами (просмотр, переключение, добавление, удаление) - ⭐ Избранные namespaces для быстрого доступа - 🎨 Красивый вывод с использованием Rich - 📦 Управление namespaces, deployments, pods, ConfigMaps @@ -13,6 +13,7 @@ - 📝 Просмотр логов pods - 🔍 Просмотр и редактирование ConfigMaps в текстовом редакторе - 💾 Сохранение настроек в `~/.config/k8s-tool/k8s-tool.cfg` +- 🔒 Автоматическое создание резервных копий kubeconfig при изменениях ## Требования @@ -58,14 +59,29 @@ poetry run k8s-tool ### Основные функции -#### Переключение контекста Kubernetes -Быстрое переключение между различными Kubernetes кластерами/контекстами: +#### Управление контекстами Kubernetes +Полное управление Kubernetes контекстами/кластерами: + +**Просмотр и переключение:** - Просмотр всех доступных контекстов в виде таблицы - Отображение текущего активного контекста (✓) - Переключение на другой контекст - Автоматическая переинициализация подключения к новому кластеру - После переключения контекста namespace сбрасывается и нужно выбрать заново +**Добавление нового контекста:** +- Интерактивное создание нового контекста +- Указание имени контекста, кластера, пользователя +- API server URL +- Опционально: токен аутентификации +- Опционально: сертификат CA (base64) +- Автоматическое создание резервной копии kubeconfig перед изменением + +**Удаление контекста:** +- Удаление ненужных контекстов +- Защита от удаления текущего активного контекста +- Автоматическое создание резервной копии kubeconfig + #### Выбор Namespace Выберите namespace для работы. Все последующие операции будут выполняться в выбранном namespace. diff --git a/k8s_tool/k8s_client.py b/k8s_tool/k8s_client.py index b109ee3..bca4b70 100644 --- a/k8s_tool/k8s_client.py +++ b/k8s_tool/k8s_client.py @@ -1,6 +1,9 @@ """Kubernetes API client wrapper.""" +import os +import yaml from typing import List, Dict, Any, Optional +from pathlib import Path from kubernetes import client, config from kubernetes.client.exceptions import ApiException from rich.console import Console @@ -65,6 +68,144 @@ class K8sClient: console.print(f"[red]Error switching context:[/red] {e}") return False + def _get_kubeconfig_path(self) -> Path: + """Get path to kubeconfig file.""" + kubeconfig = os.environ.get('KUBECONFIG') + if kubeconfig: + return Path(kubeconfig) + return Path.home() / '.kube' / 'config' + + def _load_kubeconfig(self) -> Optional[Dict]: + """Load kubeconfig file.""" + try: + kubeconfig_path = self._get_kubeconfig_path() + if not kubeconfig_path.exists(): + console.print(f"[red]Kubeconfig not found:[/red] {kubeconfig_path}") + return None + + with open(kubeconfig_path, 'r') as f: + return yaml.safe_load(f) + except Exception as e: + console.print(f"[red]Error loading kubeconfig:[/red] {e}") + return None + + def _save_kubeconfig(self, kubeconfig: Dict) -> bool: + """Save kubeconfig file.""" + try: + kubeconfig_path = self._get_kubeconfig_path() + # Create backup + backup_path = kubeconfig_path.with_suffix('.backup') + if kubeconfig_path.exists(): + import shutil + shutil.copy(kubeconfig_path, backup_path) + + with open(kubeconfig_path, 'w') as f: + yaml.dump(kubeconfig, f, default_flow_style=False) + + return True + except Exception as e: + console.print(f"[red]Error saving kubeconfig:[/red] {e}") + return False + + def delete_context(self, context_name: str) -> bool: + """Delete a context from kubeconfig.""" + try: + kubeconfig = self._load_kubeconfig() + if not kubeconfig: + return False + + # Check if context exists + contexts = kubeconfig.get('contexts', []) + context_names = [ctx['name'] for ctx in contexts] + + if context_name not in context_names: + console.print(f"[yellow]Context '{context_name}' not found[/yellow]") + return False + + # Don't allow deleting current context + if context_name == self.current_context: + console.print(f"[yellow]Cannot delete current context. Switch to another context first.[/yellow]") + return False + + # Remove context + kubeconfig['contexts'] = [ctx for ctx in contexts if ctx['name'] != context_name] + + # Save updated config + if self._save_kubeconfig(kubeconfig): + console.print(f"[green]✓[/green] Context '{context_name}' deleted") + return True + + return False + except Exception as e: + console.print(f"[red]Error deleting context:[/red] {e}") + return False + + def add_context(self, context_name: str, cluster_name: str, user_name: str, + server: str, certificate_authority: Optional[str] = None, + token: Optional[str] = None) -> bool: + """Add a new context to kubeconfig.""" + try: + kubeconfig = self._load_kubeconfig() + if not kubeconfig: + return False + + # Check if context already exists + contexts = kubeconfig.get('contexts', []) + if any(ctx['name'] == context_name for ctx in contexts): + console.print(f"[yellow]Context '{context_name}' already exists[/yellow]") + return False + + # Add cluster if it doesn't exist + clusters = kubeconfig.get('clusters', []) + if not any(cls['name'] == cluster_name for cls in clusters): + new_cluster = { + 'name': cluster_name, + 'cluster': { + 'server': server + } + } + if certificate_authority: + new_cluster['cluster']['certificate-authority-data'] = certificate_authority + else: + new_cluster['cluster']['insecure-skip-tls-verify'] = True + + clusters.append(new_cluster) + kubeconfig['clusters'] = clusters + + # Add user if it doesn't exist + users = kubeconfig.get('users', []) + if not any(usr['name'] == user_name for usr in users): + new_user = { + 'name': user_name, + 'user': {} + } + if token: + new_user['user']['token'] = token + + users.append(new_user) + kubeconfig['users'] = users + + # Add context + new_context = { + 'name': context_name, + 'context': { + 'cluster': cluster_name, + 'user': user_name + } + } + contexts.append(new_context) + kubeconfig['contexts'] = contexts + + # Save updated config + if self._save_kubeconfig(kubeconfig): + console.print(f"[green]✓[/green] Context '{context_name}' added") + return True + + return False + except Exception as e: + console.print(f"[red]Error adding context:[/red] {e}") + return False + def get_namespaces(self) -> List[str]: """Get list of all namespaces.""" try: diff --git a/k8s_tool/main.py b/k8s_tool/main.py index 0ee5a72..11d1079 100644 --- a/k8s_tool/main.py +++ b/k8s_tool/main.py @@ -74,7 +74,7 @@ class K8sTool: console.print(f"[bold]Current namespace:[/bold] {namespace_info}{fav_indicator}") choices = [ - "Switch Context", + "Manage Contexts", "Select Namespace", "Manage Favorites", "List Deployments", @@ -92,8 +92,8 @@ class K8sTool: style=custom_style ).ask() - if action == "Switch Context": - self._switch_context() + if action == "Manage Contexts": + self._manage_contexts() elif action == "Select Namespace": self._select_namespace() elif action == "Manage Favorites": @@ -152,6 +152,45 @@ class K8sTool: fav_text = " (favorite)" if is_fav else "" console.print(f"[green]✓[/green] Namespace set to: [cyan]{namespace}[/cyan]{fav_text}") + def _manage_contexts(self): + """Manage Kubernetes contexts.""" + choices = [ + "View all contexts", + "Switch context", + "Add new context", + "Delete context", + "Back to main menu" + ] + + action = questionary.select( + "Context management:", + choices=choices, + style=custom_style + ).ask() + + if not action or action == "Back to main menu": + return + + if action == "View all contexts": + self._view_contexts() + elif action == "Switch context": + self._switch_context() + elif action == "Add new context": + self._add_context() + elif action == "Delete context": + self._delete_context() + + def _view_contexts(self): + """View all available contexts.""" + console.print("[dim]Fetching contexts...[/dim]") + contexts = self.k8s_client.get_contexts() + + if not contexts: + console.print("[red]No contexts found[/red]") + return + + self.k8s_client.display_contexts_table(contexts) + def _switch_context(self): """Switch Kubernetes context.""" console.print("[dim]Fetching contexts...[/dim]") @@ -171,7 +210,7 @@ class K8sTool: console.print("[yellow]Only one context available[/yellow]") return - # Add option to view current + # Add option to cancel choices = available_contexts + [questionary.Separator(), "Cancel"] selected = questionary.select( @@ -186,6 +225,117 @@ class K8sTool: self.current_namespace = None console.print("[yellow]Note:[/yellow] Namespace selection reset. Please select a namespace.") + def _add_context(self): + """Add a new Kubernetes context.""" + console.print("\n[bold]Add New Context[/bold]\n") + + # Get context details + context_name = questionary.text( + "Context name:", + style=custom_style + ).ask() + + if not context_name: + return + + cluster_name = questionary.text( + "Cluster name:", + default=context_name, + style=custom_style + ).ask() + + if not cluster_name: + return + + server = questionary.text( + "API server URL (e.g., https://kubernetes.example.com:6443):", + style=custom_style + ).ask() + + if not server: + return + + user_name = questionary.text( + "User name:", + default=context_name + "-user", + style=custom_style + ).ask() + + if not user_name: + return + + # Optional: token + has_token = questionary.confirm( + "Do you have an authentication token?", + style=custom_style, + default=False + ).ask() + + token = None + if has_token: + token = questionary.password( + "Authentication token:", + style=custom_style + ).ask() + + # Optional: certificate + has_cert = questionary.confirm( + "Do you have a certificate authority data?", + style=custom_style, + default=False + ).ask() + + cert = None + if has_cert: + cert = questionary.text( + "Certificate authority data (base64 encoded):", + style=custom_style + ).ask() + + # Confirm + console.print("\n[bold]Review:[/bold]") + console.print(f" Context: [cyan]{context_name}[/cyan]") + console.print(f" Cluster: [cyan]{cluster_name}[/cyan]") + console.print(f" Server: [cyan]{server}[/cyan]") + console.print(f" User: [cyan]{user_name}[/cyan]") + console.print(f" Token: [cyan]{'***' if token else 'None'}[/cyan]") + console.print(f" Certificate: [cyan]{'***' if cert else 'None (insecure)'}[/cyan]") + + if questionary.confirm("\nAdd this context?", style=custom_style, default=False).ask(): + self.k8s_client.add_context(context_name, cluster_name, user_name, server, cert, token) + + def _delete_context(self): + """Delete a Kubernetes context.""" + console.print("[dim]Fetching contexts...[/dim]") + contexts = self.k8s_client.get_contexts() + + if not contexts: + console.print("[red]No contexts found[/red]") + return + + # Filter out current context + deletable_contexts = [ctx['name'] for ctx in contexts if not ctx['is_active']] + + if not deletable_contexts: + console.print("[yellow]No contexts available for deletion (cannot delete current context)[/yellow]") + return + + context_name = questionary.select( + "Select context to delete:", + choices=deletable_contexts + [questionary.Separator(), "Cancel"], + style=custom_style + ).ask() + + if not context_name or context_name == "Cancel": + return + + if questionary.confirm( + f"Are you sure you want to delete context '{context_name}'?", + style=custom_style, + default=False + ).ask(): + self.k8s_client.delete_context(context_name) + def _manage_favorites(self): """Manage favorite namespaces.""" choices = []