DevOpsLab/scripts/portforward.py

#!/usr/bin/env python3
"""
Скрипт для управления port-forward для Kubernetes сервисов
Автор: Сергей Антропов
Сайт: https://devops.org.ru
"""
import sys
import yaml
import subprocess
import os
import signal
import time

def get_cluster_name():
    """Получаем имя кластера из preset файла"""
    preset_file = "molecule/presets/k8s/kubernetes.yml"
    with open(preset_file, 'r') as f:
        preset = yaml.safe_load(f)
    return preset['kind_clusters'][0]['name']

def run_cmd(cmd):
    """Выполняет команду и возвращает результат"""
    print(f"[run] {cmd}")
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"[error] {result.stderr}")
    else:
        print(result.stdout)
    return result

def get_portforward_pids():
    """Получает PID процессов port-forward"""
    result = subprocess.run("ps aux | grep 'kubectl.*port-forward' | grep -v grep", shell=True, capture_output=True, text=True)
    pids = []
    for line in result.stdout.split('\n'):
        if line.strip():
            pids.append(int(line.split()[1]))
    return pids

def list_portforwards():
    """Показывает список всех активных port-forward"""
    pids = get_portforward_pids()
    if not pids:
        print("❌ Нет активных port-forward")
        return
    
    print("📋 Активные port-forward:")
    result = subprocess.run("ps aux | grep 'kubectl.*port-forward' | grep -v grep", shell=True, capture_output=True, text=True)
    for line in result.stdout.split('\n'):
        if line.strip():
            print(f"  {line}")

def clear_portforwards():
    """Завершает все процессы port-forward"""
    pids = get_portforward_pids()
    if not pids:
        print("❌ Нет активных port-forward")
        return
    
    print(f"🗑️  Завершение {len(pids)} процессов port-forward...")
    for pid in pids:
        try:
            os.kill(pid, signal.SIGTERM)
            print(f"✅ Процесс {pid} завершен")
        except ProcessLookupError:
            print(f"⚠️  Процесс {pid} уже не существует")
    
    # Ждем завершения процессов
    time.sleep(2)
    
    # Принудительно убиваем оставшиеся
    remaining_pids = get_portforward_pids()
    if remaining_pids:
        print("⚠️  Принудительное завершение оставшихся процессов...")
        for pid in remaining_pids:
            try:
                os.kill(pid, signal.SIGKILL)
                print(f"✅ Процесс {pid} принудительно завершен")
            except ProcessLookupError:
                pass

def create_portforwards():
    """Создает port-forward для всех сервисов из preset"""
    # Загружаем preset
    preset_file = "molecule/presets/k8s/kubernetes.yml"
    with open(preset_file, 'r') as f:
        preset = yaml.safe_load(f)
    
    cluster_name = preset['kind_clusters'][0]['name']
    addon_ports = preset['kind_clusters'][0].get('addon_ports', {})
    
    print(f"🔌 Создание port-forward для кластера: {cluster_name}")
    
    # Prometheus
    if addon_ports.get('prometheus'):
        port = addon_ports['prometheus']
        print(f"  - Prometheus: localhost:{port} -> monitoring/monitoring-kube-prometheus-prometheus:9090")
        subprocess.Popen([
            "kubectl", 
            "--server=https://{}-control-plane:6443".format(cluster_name),
            "--insecure-skip-tls-verify",
            "port-forward",
            "-n", "monitoring",
            "svc/monitoring-kube-prometheus-prometheus",
            f"{port}:9090"
        ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    
    # Grafana
    if addon_ports.get('grafana'):
        port = addon_ports['grafana']
        print(f"  - Grafana: localhost:{port} -> monitoring/monitoring-grafana:80")
        subprocess.Popen([
            "kubectl",
            "--server=https://{}-control-plane:6443".format(cluster_name),
            "--insecure-skip-tls-verify",
            "port-forward",
            "-n", "monitoring",
            "svc/monitoring-grafana",
            f"{port}:80"
        ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    
    # Kiali
    if addon_ports.get('kiali'):
        port = addon_ports['kiali']
        print(f"  - Kiali: localhost:{port} -> istio-system/kiali:20001")
        subprocess.Popen([
            "kubectl",
            "--server=https://{}-control-plane:6443".format(cluster_name),
            "--insecure-skip-tls-verify",
            "port-forward",
            "-n", "istio-system",
            "svc/kiali",
            f"{port}:20001"
        ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    
    # Metrics Server
    if addon_ports.get('metrics_server'):
        port = addon_ports['metrics_server']
        print(f"  - Metrics Server: localhost:{port} -> kube-system/metrics-server:4443")
        subprocess.Popen([
            "kubectl",
            "--server=https://{}-control-plane:6443".format(cluster_name),
            "--insecure-skip-tls-verify",
            "port-forward",
            "-n", "kube-system",
            "svc/metrics-server",
            f"{port}:4443"
        ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    
    time.sleep(2)
    print("✅ Port-forward создан")
    list_portforwards()

def delete_portforward(port):
    """Удаляет port-forward для конкретного порта"""
    pids = get_portforward_pids()
    if not pids:
        print(f"❌ Нет активных port-forward на порту {port}")
        return
    
    # Находим процесс с нужным портом
    result = subprocess.run(f"ps aux | grep 'kubectl.*port-forward.*:{port}' | grep -v grep", shell=True, capture_output=True, text=True)
    if not result.stdout.strip():
        print(f"❌ Не найден port-forward на порту {port}")
        return
    
    # Извлекаем PID
    pid = int(result.stdout.split()[1])
    print(f"🗑️  Завершение port-forward на порту {port} (PID: {pid})...")
    try:
        os.kill(pid, signal.SIGTERM)
        print(f"✅ Port-forward на порту {port} завершен")
    except ProcessLookupError:
        print(f"⚠️  Процесс {pid} уже не существует")

def main():
    if len(sys.argv) < 2:
        print("Usage: portforward.py <create|list|delete|clear> [port]")
        sys.exit(1)
    
    command = sys.argv[1]
    
    if command == "create":
        create_portforwards()
    elif command == "list":
        list_portforwards()
    elif command == "clear":
        clear_portforwards()
    elif command == "delete":
        if len(sys.argv) < 3:
            print("Usage: portforward.py delete <port>")
            sys.exit(1)
        port = sys.argv[2]
        delete_portforward(port)
    else:
        print(f"❌ Неизвестная команда: {command}")
        sys.exit(1)

if __name__ == "__main__":
    main()