bvn13/k8s-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working setup

Browse Source
This commit is contained in:
vy.boyko 2025-10-25 20:59:29 +03:00
commit b76941ec3a
6 changed files with 849 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
.gitignore vendored Normal file
View File

@ -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

131
README.md Normal file
View File

@ -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 <repo-url>
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

3
k8s_tool/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""K8s Tool - Interactive kubectl helper."""
__version__ = "0.1.0"

219
k8s_tool/k8s_client.py Normal file
View File

@ -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)

335
k8s_tool/main.py Normal file
View File

@ -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()

21
pyproject.toml Normal file
View File

@ -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"