From b76941ec3afd59e98a6dcff5be9b17a06c049036 Mon Sep 17 00:00:00 2001 From: "vy.boyko" Date: Sat, 25 Oct 2025 20:59:29 +0300 Subject: [PATCH] working setup --- .gitignore | 140 +++++++++++++++++ README.md | 131 ++++++++++++++++ k8s_tool/__init__.py | 3 + k8s_tool/k8s_client.py | 219 +++++++++++++++++++++++++++ k8s_tool/main.py | 335 +++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 21 +++ 6 files changed, 849 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 k8s_tool/__init__.py create mode 100644 k8s_tool/k8s_client.py create mode 100644 k8s_tool/main.py create mode 100644 pyproject.toml diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..84d5996 --- /dev/null +++ b/.gitignore @@ -0,0 +1,140 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +Pipfile.lock + +# Poetry +poetry.lock + +# PEP 582 +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# IDEs +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + + +.claude \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..43672c3 --- /dev/null +++ b/README.md @@ -0,0 +1,131 @@ +# K8s Tool + +Интерактивное консольное приложение для упрощения работы с kubectl. Предоставляет удобное меню с навигацией стрелками для выполнения рутинных операций с Kubernetes. + +## Особенности + +- 🎯 Интерактивное меню с навигацией стрелками +- 🎨 Красивый вывод с использованием Rich +- 📦 Управление namespaces, deployments, pods, ConfigMaps +- 🔄 Быстрый рестарт и масштабирование deployments +- 📝 Просмотр логов pods +- 🔍 Просмотр содержимого ConfigMaps + +## Требования + +- Python 3.9+ +- Poetry +- kubectl настроенный для работы с кластером +- Доступ к Kubernetes кластеру + +## Установка + +1. Клонируйте репозиторий: +```bash +git clone +cd k8s-tool +``` + +2. Установите зависимости с помощью Poetry: +```bash +poetry install +``` + +3. Активируйте виртуальное окружение: +```bash +poetry shell +``` + +4. Установите приложение: +```bash +poetry install +``` + +## Использование + +Запустите приложение командой: +```bash +k8s-tool +``` + +Или через Poetry: +```bash +poetry run k8s-tool +``` + +### Основные функции + +#### Выбор Namespace +Выберите namespace для работы. Все последующие операции будут выполняться в выбранном namespace. + +#### Список Deployments +Отображает все deployments в текущем namespace с информацией о: +- Имени deployment +- Количестве реплик +- Доступных репликах +- Готовых репликах + +Можно также просмотреть список pods для конкретного deployment. + +#### Рестарт Deployment +Перезапускает выбранный deployment путём обновления аннотации `kubectl.kubernetes.io/restartedAt`. + +#### Масштабирование Deployment +Изменяет количество реплик deployment. Введите желаемое количество реплик, и приложение выполнит масштабирование. + +#### Просмотр ConfigMaps +Отображает список всех ConfigMaps в namespace с: +- Именем ConfigMap +- Списком ключей данных +- Возрастом ConfigMap + +Можно выбрать ConfigMap для просмотра его содержимого с подсветкой синтаксиса. + +#### Просмотр логов Pod +Выберите pod и контейнер (если их несколько) для просмотра логов. Можно указать количество строк для отображения (по умолчанию 100). + +## Навигация + +- Используйте стрелки ↑/↓ для перемещения по меню +- Enter для выбора опции +- Ctrl+C для выхода из приложения + +## Архитектура + +Проект состоит из двух основных модулей: + +- `k8s_client.py` - обёртка над Kubernetes Python API для работы с кластером +- `main.py` - главное приложение с интерактивным меню + +### Технологии + +- **questionary** - интерактивные промпты с навигацией стрелками +- **kubernetes** - Python client для Kubernetes API +- **rich** - красивый вывод в терминале с таблицами и подсветкой синтаксиса + +## Разработка + +Для разработки: + +1. Установите зависимости: +```bash +poetry install +``` + +2. Активируйте виртуальное окружение: +```bash +poetry shell +``` + +3. Запустите приложение: +```bash +python -m k8s_tool.main +``` + +## Лицензия + +MIT + +## Автор + +vy.boyko diff --git a/k8s_tool/__init__.py b/k8s_tool/__init__.py new file mode 100644 index 0000000..a9d50a1 --- /dev/null +++ b/k8s_tool/__init__.py @@ -0,0 +1,3 @@ +"""K8s Tool - Interactive kubectl helper.""" + +__version__ = "0.1.0" diff --git a/k8s_tool/k8s_client.py b/k8s_tool/k8s_client.py new file mode 100644 index 0000000..434f72a --- /dev/null +++ b/k8s_tool/k8s_client.py @@ -0,0 +1,219 @@ +"""Kubernetes API client wrapper.""" + +from typing import List, Dict, Any, Optional +from kubernetes import client, config +from kubernetes.client.exceptions import ApiException +from rich.console import Console +from rich.table import Table +from datetime import datetime + +console = Console() + + +class K8sClient: + """Wrapper for Kubernetes API operations.""" + + def __init__(self): + """Initialize Kubernetes client.""" + try: + config.load_kube_config() + self.v1 = client.CoreV1Api() + self.apps_v1 = client.AppsV1Api() + console.print("[green]✓[/green] Connected to Kubernetes cluster") + except Exception as e: + console.print(f"[red]✗[/red] Failed to connect to Kubernetes: {e}") + raise + + def get_namespaces(self) -> List[str]: + """Get list of all namespaces.""" + try: + namespaces = self.v1.list_namespace() + return sorted([ns.metadata.name for ns in namespaces.items]) + except ApiException as e: + console.print(f"[red]Error fetching namespaces:[/red] {e}") + return [] + + def get_deployments(self, namespace: str) -> List[Dict[str, Any]]: + """Get list of deployments in namespace.""" + try: + deployments = self.apps_v1.list_namespaced_deployment(namespace) + return [{ + 'name': dep.metadata.name, + 'replicas': dep.spec.replicas, + 'available': dep.status.available_replicas or 0, + 'ready': dep.status.ready_replicas or 0, + } for dep in deployments.items] + except ApiException as e: + console.print(f"[red]Error fetching deployments:[/red] {e}") + return [] + + def get_pods(self, namespace: str, label_selector: Optional[str] = None) -> List[Dict[str, Any]]: + """Get list of pods in namespace.""" + try: + if label_selector: + pods = self.v1.list_namespaced_pod(namespace, label_selector=label_selector) + else: + pods = self.v1.list_namespaced_pod(namespace) + + return [{ + 'name': pod.metadata.name, + 'status': pod.status.phase, + 'ready': sum(1 for c in pod.status.container_statuses or [] if c.ready), + 'total_containers': len(pod.spec.containers), + 'restarts': sum(c.restart_count for c in pod.status.container_statuses or []), + 'age': self._calculate_age(pod.metadata.creation_timestamp), + } for pod in pods.items] + except ApiException as e: + console.print(f"[red]Error fetching pods:[/red] {e}") + return [] + + def get_configmaps(self, namespace: str) -> List[Dict[str, Any]]: + """Get list of ConfigMaps in namespace.""" + try: + configmaps = self.v1.list_namespaced_config_map(namespace) + return [{ + 'name': cm.metadata.name, + 'data_keys': list(cm.data.keys()) if cm.data else [], + 'age': self._calculate_age(cm.metadata.creation_timestamp), + } for cm in configmaps.items] + except ApiException as e: + console.print(f"[red]Error fetching ConfigMaps:[/red] {e}") + return [] + + def get_configmap_data(self, namespace: str, name: str) -> Optional[Dict[str, str]]: + """Get ConfigMap data.""" + try: + cm = self.v1.read_namespaced_config_map(name, namespace) + return cm.data + except ApiException as e: + console.print(f"[red]Error reading ConfigMap:[/red] {e}") + return None + + def restart_deployment(self, namespace: str, name: str) -> bool: + """Restart deployment by updating annotation.""" + try: + now = datetime.utcnow().isoformat() + "Z" + body = { + 'spec': { + 'template': { + 'metadata': { + 'annotations': { + 'kubectl.kubernetes.io/restartedAt': now + } + } + } + } + } + self.apps_v1.patch_namespaced_deployment(name, namespace, body) + console.print(f"[green]✓[/green] Deployment {name} restarted") + return True + except ApiException as e: + console.print(f"[red]Error restarting deployment:[/red] {e}") + return False + + def scale_deployment(self, namespace: str, name: str, replicas: int) -> bool: + """Scale deployment to specified number of replicas.""" + try: + body = {'spec': {'replicas': replicas}} + self.apps_v1.patch_namespaced_deployment_scale(name, namespace, body) + console.print(f"[green]✓[/green] Deployment {name} scaled to {replicas} replicas") + return True + except ApiException as e: + console.print(f"[red]Error scaling deployment:[/red] {e}") + return False + + def get_pod_logs(self, namespace: str, pod_name: str, container: Optional[str] = None, + tail_lines: int = 100) -> Optional[str]: + """Get logs from pod.""" + try: + kwargs = {'name': pod_name, 'namespace': namespace, 'tail_lines': tail_lines} + if container: + kwargs['container'] = container + + logs = self.v1.read_namespaced_pod_log(**kwargs) + return logs + except ApiException as e: + console.print(f"[red]Error fetching logs:[/red] {e}") + return None + + def get_pod_containers(self, namespace: str, pod_name: str) -> List[str]: + """Get list of containers in pod.""" + try: + pod = self.v1.read_namespaced_pod(pod_name, namespace) + return [container.name for container in pod.spec.containers] + except ApiException as e: + console.print(f"[red]Error fetching pod info:[/red] {e}") + return [] + + @staticmethod + def _calculate_age(timestamp: datetime) -> str: + """Calculate age from timestamp.""" + if not timestamp: + return "Unknown" + + now = datetime.now(timestamp.tzinfo) + delta = now - timestamp + + if delta.days > 0: + return f"{delta.days}d" + elif delta.seconds >= 3600: + return f"{delta.seconds // 3600}h" + elif delta.seconds >= 60: + return f"{delta.seconds // 60}m" + else: + return f"{delta.seconds}s" + + def display_deployments_table(self, deployments: List[Dict[str, Any]]): + """Display deployments in a table.""" + table = Table(title="Deployments") + table.add_column("Name", style="cyan") + table.add_column("Replicas", style="magenta") + table.add_column("Available", style="green") + table.add_column("Ready", style="yellow") + + for dep in deployments: + table.add_row( + dep['name'], + str(dep['replicas']), + str(dep['available']), + str(dep['ready']) + ) + + console.print(table) + + def display_pods_table(self, pods: List[Dict[str, Any]]): + """Display pods in a table.""" + table = Table(title="Pods") + table.add_column("Name", style="cyan") + table.add_column("Status", style="magenta") + table.add_column("Ready", style="green") + table.add_column("Restarts", style="yellow") + table.add_column("Age", style="blue") + + for pod in pods: + status_color = "green" if pod['status'] == "Running" else "red" + table.add_row( + pod['name'], + f"[{status_color}]{pod['status']}[/{status_color}]", + f"{pod['ready']}/{pod['total_containers']}", + str(pod['restarts']), + pod['age'] + ) + + console.print(table) + + def display_configmaps_table(self, configmaps: List[Dict[str, Any]]): + """Display ConfigMaps in a table.""" + table = Table(title="ConfigMaps") + table.add_column("Name", style="cyan") + table.add_column("Keys", style="magenta") + table.add_column("Age", style="blue") + + for cm in configmaps: + table.add_row( + cm['name'], + ", ".join(cm['data_keys']) if cm['data_keys'] else "None", + cm['age'] + ) + + console.print(table) diff --git a/k8s_tool/main.py b/k8s_tool/main.py new file mode 100644 index 0000000..463dc9f --- /dev/null +++ b/k8s_tool/main.py @@ -0,0 +1,335 @@ +"""Main application entry point.""" + +import sys +from typing import Optional +import questionary +from questionary import Style +from rich.console import Console +from rich.panel import Panel +from rich.syntax import Syntax + +from k8s_tool.k8s_client import K8sClient + +console = Console() + +# Custom style for questionary prompts +custom_style = Style([ + ('qmark', 'fg:#673ab7 bold'), + ('question', 'bold'), + ('answer', 'fg:#f44336 bold'), + ('pointer', 'fg:#673ab7 bold'), + ('highlighted', 'fg:#673ab7 bold'), + ('selected', 'fg:#cc5454'), + ('separator', 'fg:#cc5454'), + ('instruction', ''), + ('text', ''), +]) + + +class K8sTool: + """Main application class.""" + + def __init__(self): + """Initialize the application.""" + self.k8s_client = K8sClient() + self.current_namespace: Optional[str] = None + + def run(self): + """Run the main application loop.""" + console.print(Panel.fit( + "[bold cyan]K8s Tool[/bold cyan]\n" + "[dim]Interactive kubectl helper[/dim]", + border_style="cyan" + )) + + while True: + try: + self._main_menu() + except KeyboardInterrupt: + console.print("\n[yellow]Goodbye! 👋[/yellow]") + sys.exit(0) + except Exception as e: + console.print(f"[red]Error:[/red] {e}") + if questionary.confirm("Continue?", style=custom_style).ask(): + continue + else: + break + + def _main_menu(self): + """Display main menu.""" + namespace_info = f"[cyan]{self.current_namespace}[/cyan]" if self.current_namespace else "[dim]not selected[/dim]" + console.print(f"\n[bold]Current namespace:[/bold] {namespace_info}") + + choices = [ + "Select Namespace", + "List Deployments", + "Restart Deployment", + "Scale Deployment", + "View ConfigMaps", + "View Pod Logs", + "Exit" + ] + + action = questionary.select( + "What would you like to do?", + choices=choices, + style=custom_style + ).ask() + + if action == "Select Namespace": + self._select_namespace() + elif action == "List Deployments": + self._list_deployments() + elif action == "Restart Deployment": + self._restart_deployment() + elif action == "Scale Deployment": + self._scale_deployment() + elif action == "View ConfigMaps": + self._view_configmaps() + elif action == "View Pod Logs": + self._view_pod_logs() + elif action == "Exit": + console.print("[yellow]Goodbye! 👋[/yellow]") + sys.exit(0) + + def _select_namespace(self): + """Select namespace.""" + console.print("[dim]Fetching namespaces...[/dim]") + namespaces = self.k8s_client.get_namespaces() + + if not namespaces: + console.print("[red]No namespaces found[/red]") + return + + namespace = questionary.select( + "Select namespace:", + choices=namespaces, + style=custom_style + ).ask() + + if namespace: + self.current_namespace = namespace + console.print(f"[green]✓[/green] Namespace set to: [cyan]{namespace}[/cyan]") + + def _ensure_namespace_selected(self) -> bool: + """Ensure namespace is selected.""" + if not self.current_namespace: + console.print("[yellow]Please select a namespace first[/yellow]") + self._select_namespace() + return self.current_namespace is not None + return True + + def _list_deployments(self): + """List deployments in current namespace.""" + if not self._ensure_namespace_selected(): + return + + console.print(f"[dim]Fetching deployments in {self.current_namespace}...[/dim]") + deployments = self.k8s_client.get_deployments(self.current_namespace) + + if not deployments: + console.print("[yellow]No deployments found[/yellow]") + return + + self.k8s_client.display_deployments_table(deployments) + + if questionary.confirm("View pods for a deployment?", style=custom_style, default=False).ask(): + dep_names = [dep['name'] for dep in deployments] + dep_name = questionary.select( + "Select deployment:", + choices=dep_names, + style=custom_style + ).ask() + + if dep_name: + self._view_deployment_pods(dep_name) + + def _view_deployment_pods(self, deployment_name: str): + """View pods for a deployment.""" + label_selector = f"app={deployment_name}" + console.print(f"[dim]Fetching pods for deployment {deployment_name}...[/dim]") + pods = self.k8s_client.get_pods(self.current_namespace, label_selector=label_selector) + + if not pods: + # Try without label selector + console.print("[yellow]No pods found with label selector, fetching all pods...[/yellow]") + pods = self.k8s_client.get_pods(self.current_namespace) + + if pods: + self.k8s_client.display_pods_table(pods) + else: + console.print("[yellow]No pods found[/yellow]") + + def _restart_deployment(self): + """Restart a deployment.""" + if not self._ensure_namespace_selected(): + return + + console.print(f"[dim]Fetching deployments in {self.current_namespace}...[/dim]") + deployments = self.k8s_client.get_deployments(self.current_namespace) + + if not deployments: + console.print("[yellow]No deployments found[/yellow]") + return + + dep_names = [dep['name'] for dep in deployments] + dep_name = questionary.select( + "Select deployment to restart:", + choices=dep_names, + style=custom_style + ).ask() + + if not dep_name: + return + + confirm = questionary.confirm( + f"Are you sure you want to restart {dep_name}?", + style=custom_style, + default=False + ).ask() + + if confirm: + self.k8s_client.restart_deployment(self.current_namespace, dep_name) + + def _scale_deployment(self): + """Scale a deployment.""" + if not self._ensure_namespace_selected(): + return + + console.print(f"[dim]Fetching deployments in {self.current_namespace}...[/dim]") + deployments = self.k8s_client.get_deployments(self.current_namespace) + + if not deployments: + console.print("[yellow]No deployments found[/yellow]") + return + + dep_names = [f"{dep['name']} (current: {dep['replicas']})" for dep in deployments] + selected = questionary.select( + "Select deployment to scale:", + choices=dep_names, + style=custom_style + ).ask() + + if not selected: + return + + dep_name = selected.split(" (current:")[0] + + replicas = questionary.text( + "Enter number of replicas:", + validate=lambda text: text.isdigit() and int(text) >= 0, + style=custom_style + ).ask() + + if replicas: + self.k8s_client.scale_deployment(self.current_namespace, dep_name, int(replicas)) + + def _view_configmaps(self): + """View ConfigMaps.""" + if not self._ensure_namespace_selected(): + return + + console.print(f"[dim]Fetching ConfigMaps in {self.current_namespace}...[/dim]") + configmaps = self.k8s_client.get_configmaps(self.current_namespace) + + if not configmaps: + console.print("[yellow]No ConfigMaps found[/yellow]") + return + + self.k8s_client.display_configmaps_table(configmaps) + + if questionary.confirm("View ConfigMap data?", style=custom_style, default=False).ask(): + cm_names = [cm['name'] for cm in configmaps] + cm_name = questionary.select( + "Select ConfigMap:", + choices=cm_names, + style=custom_style + ).ask() + + if cm_name: + data = self.k8s_client.get_configmap_data(self.current_namespace, cm_name) + if data: + console.print(f"\n[bold]ConfigMap:[/bold] [cyan]{cm_name}[/cyan]\n") + for key, value in data.items(): + console.print(Panel( + Syntax(value, "yaml", theme="monokai", line_numbers=True), + title=f"[cyan]{key}[/cyan]", + border_style="cyan" + )) + + def _view_pod_logs(self): + """View pod logs.""" + if not self._ensure_namespace_selected(): + return + + console.print(f"[dim]Fetching pods in {self.current_namespace}...[/dim]") + pods = self.k8s_client.get_pods(self.current_namespace) + + if not pods: + console.print("[yellow]No pods found[/yellow]") + return + + pod_names = [f"{pod['name']} ({pod['status']})" for pod in pods] + selected = questionary.select( + "Select pod:", + choices=pod_names, + style=custom_style + ).ask() + + if not selected: + return + + pod_name = selected.split(" (")[0] + + # Check if pod has multiple containers + containers = self.k8s_client.get_pod_containers(self.current_namespace, pod_name) + container = None + + if len(containers) > 1: + container = questionary.select( + "Select container:", + choices=containers, + style=custom_style + ).ask() + elif len(containers) == 1: + container = containers[0] + + tail_lines = questionary.text( + "Number of lines to show (default: 100):", + default="100", + validate=lambda text: text.isdigit() and int(text) > 0, + style=custom_style + ).ask() + + if tail_lines: + logs = self.k8s_client.get_pod_logs( + self.current_namespace, + pod_name, + container=container, + tail_lines=int(tail_lines) + ) + + if logs: + console.print(Panel( + Syntax(logs, "log", theme="monokai", line_numbers=False), + title=f"[cyan]Logs: {pod_name}" + (f" [{container}]" if container else "") + "[/cyan]", + border_style="cyan", + expand=False + )) + + +def main(): + """Main entry point.""" + try: + app = K8sTool() + app.run() + except KeyboardInterrupt: + console.print("\n[yellow]Goodbye! 👋[/yellow]") + sys.exit(0) + except Exception as e: + console.print(f"[red]Fatal error:[/red] {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..debda16 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "k8s-tool" +version = "0.1.0" +description = "Interactive kubectl helper tool" +authors = [ + {name = "vy.boyko"} +] +readme = "README.md" +requires-python = "^3.9" +dependencies = [ + "questionary (>=2.1.1,<3.0.0)", + "kubernetes (>=34.1.0,<35.0.0)", + "rich (>=14.2.0,<15.0.0)" +] + +[project.scripts] +k8s-tool = "k8s_tool.main:main" + +[build-system] +requires = ["poetry-core>=2.0.0,<3.0.0"] +build-backend = "poetry.core.masonry.api"