Jiří Stachandaider
c47bf0ea16
Co-authored-by: aider (openrouter/thinkingmachines/inkling-small:free) <aider@aider.chat>
128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import subprocess
|
|
|
|
app = FastAPI()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
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
|
|
|
|
@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}
|
|
|
|
@app.get("/pods")
|
|
def get_pods(namespace: str = "default", context: str = None):
|
|
result = run_kubectl(
|
|
["get", "pods", "-n", namespace, "-o", "json"],
|
|
context=context
|
|
)
|
|
return {"raw": result.stdout, "namespace": namespace, "context": context}
|
|
|
|
@app.get("/nodes")
|
|
def get_nodes(context: str = None):
|
|
result = run_kubectl(
|
|
["get", "nodes", "-o", "json"],
|
|
context=context
|
|
)
|
|
return {"raw": result.stdout, "context": context}
|
|
|
|
@app.get("/pvcs")
|
|
def get_pvcs(namespace: str = "default", context: str = None):
|
|
result = run_kubectl(
|
|
["get", "pvc", "-n", namespace, "-o", "json"],
|
|
context=context
|
|
)
|
|
return {"raw": result.stdout, "namespace": namespace, "context": context}
|
|
|
|
@app.get("/configmaps")
|
|
def get_configmaps(namespace: str = "default", context: str = None):
|
|
result = run_kubectl(
|
|
["get", "configmaps", "-n", namespace, "-o", "json"],
|
|
context=context
|
|
)
|
|
return {"raw": result.stdout, "namespace": namespace, "context": context}
|
|
|
|
@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}
|