Jiří Stachandaider
ebc8196c02
Co-authored-by: aider (openrouter/thinkingmachines/inkling-small:free) <aider@aider.chat>
241 lines
9.1 KiB
Python
241 lines
9.1 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import subprocess
|
|
import time
|
|
import json
|
|
import threading
|
|
import os
|
|
|
|
app = FastAPI()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
CACHE_FILE = "cache.json"
|
|
CACHE_TTL = 300 # 5 minutes
|
|
cache = {}
|
|
cache_lock = threading.Lock()
|
|
|
|
def load_persistent_cache():
|
|
global cache
|
|
if not os.path.exists(CACHE_FILE):
|
|
return
|
|
try:
|
|
with open(CACHE_FILE, "r") as f:
|
|
data = json.load(f)
|
|
loaded = {}
|
|
for k, v in data.items():
|
|
if isinstance(v, list) and len(v) == 2:
|
|
loaded[k] = (v[0], v[1])
|
|
with cache_lock:
|
|
cache = loaded
|
|
except Exception:
|
|
pass
|
|
|
|
def save_persistent_cache():
|
|
try:
|
|
with cache_lock:
|
|
data = {}
|
|
for k, v in cache.items():
|
|
data[k] = [v[0], v[1]]
|
|
with open(CACHE_FILE, "w") as f:
|
|
json.dump(data, f)
|
|
except Exception:
|
|
pass
|
|
|
|
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)
|
|
with cache_lock:
|
|
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)
|
|
with cache_lock:
|
|
cache[key] = (time.time(), items)
|
|
save_persistent_cache()
|
|
|
|
def refresh_worker(resource, namespace, context, kubectl_args, item_mapper):
|
|
while True:
|
|
try:
|
|
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)
|
|
except Exception:
|
|
pass
|
|
time.sleep(240) # 4 minutes background renewal
|
|
|
|
@app.on_event("startup")
|
|
def start_background_refresh():
|
|
load_persistent_cache()
|
|
workers = [
|
|
("pods", "default", None, ["get", "pods", "-n", "default", "-o", "json"], lambda i: {"name": i.get("metadata", {}).get("name", "unknown"), "namespace": i.get("metadata", {}).get("namespace", "default")}),
|
|
("nodes", "default", None, ["get", "nodes", "-o", "json"], lambda i: {"name": i.get("metadata", {}).get("name", "unknown"), "namespace": "-"}),
|
|
("pvcs", "default", None, ["get", "pvc", "-n", "default", "-o", "json"], lambda i: {"name": i.get("metadata", {}).get("name", "unknown"), "namespace": i.get("metadata", {}).get("namespace", "default")}),
|
|
("configmaps", "default", None, ["get", "configmaps", "-n", "default", "-o", "json"], lambda i: {"name": i.get("metadata", {}).get("name", "unknown"), "namespace": i.get("metadata", {}).get("namespace", "default")}),
|
|
]
|
|
for res, ns, ctx, args, mapper in workers:
|
|
t = threading.Thread(target=refresh_worker, args=(res, ns, ctx, args, mapper), daemon=True)
|
|
t.start()
|
|
|
|
@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}
|