kubectl-gui/backend/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import subprocess
import time
import json

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

cache = {}
CACHE_TTL = 300  # 5 minutes

def run_kubectl(args, context=None):
    cmd = ["kubectl"]
    if context:
        cmd += ["--context", context]
    cmd += args
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result

def get_cache_key(resource, namespace, context):
    return f"{resource}:{namespace}:{context or ''}"

def get_cached_items(resource, namespace, context):
    key = get_cache_key(resource, namespace, context)
    if key in cache:
        ts, items = cache[key]
        if time.time() - ts < CACHE_TTL:
            return items
    return None

def set_cached_items(resource, namespace, context, items):
    key = get_cache_key(resource, namespace, context)
    cache[key] = (time.time(), items)

@app.get("/")
def root():
    return {"message": "Kubectl Web GUI backend"}

@app.get("/contexts")
def get_contexts():
    result = run_kubectl(["config", "get-contexts", "-o", "name"])
    contexts = [line for line in result.stdout.splitlines() if line]
    current = ""
    try:
        cur = run_kubectl(["config", "current-context"])
        current = cur.stdout.strip()
    except Exception:
        pass
    return {"current": current, "contexts": contexts}

@app.get("/namespaces")
def get_namespaces(context: str = None):
    result = run_kubectl(
        ["get", "namespaces", "-o", "jsonpath={range .items[*]}{.metadata.name}{\"\\n\"}{end}"],
        context=context
    )
    namespaces = [line for line in result.stdout.splitlines() if line]
    return {"namespaces": namespaces}

def paginated_list(resource, namespace, context, search, page, page_size, bypass, kubectl_args, item_mapper):
    key = get_cache_key(resource, namespace, context)
    items = None
    if not bypass:
        items = get_cached_items(resource, namespace, context)
    if items is None:
        result = run_kubectl(kubectl_args, context=context)
        try:
            obj = json.loads(result.stdout)
            raw_items = obj.get("items", [])
        except Exception:
            raw_items = []
        items = [item_mapper(i) for i in raw_items]
        set_cached_items(resource, namespace, context, items)
    if search:
        s = search.lower()
        items = [i for i in items if s in (i.get("name") or "").lower()]
    total = len(items)
    start = (page - 1) * page_size
    paginated = items[start:start + page_size]
    return {
        "items": paginated,
        "total": total,
        "page": page,
        "page_size": page_size
    }

@app.get("/pods")
def get_pods(namespace: str = "default", context: str = None, search: str = "", page: int = 1, page_size: int = 10, bypass: bool = False):
    def mapper(i):
        return {
            "name": i.get("metadata", {}).get("name", "unknown"),
            "namespace": i.get("metadata", {}).get("namespace", namespace)
        }
    return paginated_list("pods", namespace, context, search, page, page_size, bypass, ["get", "pods", "-n", namespace, "-o", "json"], mapper)

@app.get("/nodes")
def get_nodes(context: str = None, search: str = "", page: int = 1, page_size: int = 10, bypass: bool = False):
    def mapper(i):
        return {
            "name": i.get("metadata", {}).get("name", "unknown"),
            "namespace": "-"
        }
    return paginated_list("nodes", "default", context, search, page, page_size, bypass, ["get", "nodes", "-o", "json"], mapper)

@app.get("/pvcs")
def get_pvcs(namespace: str = "default", context: str = None, search: str = "", page: int = 1, page_size: int = 10, bypass: bool = False):
    def mapper(i):
        return {
            "name": i.get("metadata", {}).get("name", "unknown"),
            "namespace": i.get("metadata", {}).get("namespace", namespace)
        }
    return paginated_list("pvcs", namespace, context, search, page, page_size, bypass, ["get", "pvc", "-n", namespace, "-o", "json"], mapper)

@app.get("/configmaps")
def get_configmaps(namespace: str = "default", context: str = None, search: str = "", page: int = 1, page_size: int = 10, bypass: bool = False):
    def mapper(i):
        return {
            "name": i.get("metadata", {}).get("name", "unknown"),
            "namespace": i.get("metadata", {}).get("namespace", namespace)
        }
    return paginated_list("configmaps", namespace, context, search, page, page_size, bypass, ["get", "configmaps", "-n", namespace, "-o", "json"], mapper)

@app.get("/describe")
def describe(resource: str, name: str, namespace: str = "default", context: str = None):
    allowed = {"pod", "node", "pvc", "service", "deployment", "configmap"}
    if resource not in allowed:
        return {"error": "Invalid resource type"}
    result = run_kubectl(
        ["describe", resource, name, "-n", namespace],
        context=context
    )
    return {"output": result.stdout, "stderr": result.stderr, "returncode": result.returncode}

@app.post("/delete")
def delete(resource: str, name: str, namespace: str = "default", context: str = None):
    allowed = {"pod", "node", "pvc", "service", "deployment", "configmap"}
    if resource not in allowed:
        return {"error": "Invalid resource type"}
    result = run_kubectl(
        ["delete", resource, name, "-n", namespace, "--ignore-not-found=true"],
        context=context
    )
    return {"output": result.stdout, "stderr": result.stderr, "returncode": result.returncode}

@app.get("/logs")
def logs(pod: str, namespace: str = "default", context: str = None, tail: int = 100):
    result = run_kubectl(
        ["logs", pod, "-n", namespace, "--tail", str(tail)],
        context=context
    )
    return {"output": result.stdout, "stderr": result.stderr, "returncode": result.returncode}

@app.get("/exec")
def exec_pod(pod: str, namespace: str = "default", context: str = None, container: str = None, command: str = "ls"):
    cmd = ["exec", pod, "-n", namespace]
    if container:
        cmd += ["-c", container]
    cmd += ["--", "sh", "-c", command]
    result = run_kubectl(cmd, context=context)
    return {"output": result.stdout, "stderr": result.stderr, "returncode": result.returncode}

@app.get("/cordon")
def cordon(node: str, context: str = None):
    result = run_kubectl(["cordon", node], context=context)
    return {"output": result.stdout, "stderr": result.stderr, "returncode": result.returncode}

@app.get("/uncordon")
def uncordon(node: str, context: str = None):
    result = run_kubectl(["uncordon", node], context=context)
    return {"output": result.stdout, "stderr": result.stderr, "returncode": result.returncode}