claude-code
175249ea61
mypy attr-defined errors on 'from kubernetes import config/client' in request_upgrades.py and web.py — same fix already applied to check_versions.py.
308 lines
12 KiB
Python
308 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
web.py: FastAPI web dashboard for Kubernetes image version tracking.
|
|
|
|
Routes:
|
|
GET / — image inventory grouped by namespace → app → container
|
|
GET /upgrade — upgrade trigger page with outdated images and job history
|
|
POST /api/refresh — kick off a background version-check refresh
|
|
POST /api/upgrade/trigger — create a Job from the upgrade-requester CronJob
|
|
GET /api/upgrade/jobs — recent upgrade job history (JSON)
|
|
GET /api/images — full inventory JSON
|
|
"""
|
|
|
|
import logging
|
|
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import BackgroundTasks, FastAPI, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from kubernetes import client, config # type: ignore[attr-defined]
|
|
from packaging.version import InvalidVersion, Version
|
|
|
|
from check_versions import get_latest_version, parse_image_ref
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI(title="K8s Image Dashboard")
|
|
templates = Jinja2Templates(directory="templates")
|
|
_pool = ThreadPoolExecutor(max_workers=20)
|
|
|
|
_cache: dict = {"inventory": None, "last_updated": None, "is_checking": False}
|
|
|
|
|
|
# ── Kubernetes helpers ────────────────────────────────────────────────────────
|
|
|
|
def _load_k8s() -> None:
|
|
try:
|
|
config.load_incluster_config()
|
|
except config.ConfigException:
|
|
config.load_kube_config()
|
|
|
|
|
|
def _build_owner_maps() -> tuple[dict, dict]:
|
|
apps_v1 = client.AppsV1Api()
|
|
batch_v1 = client.BatchV1Api()
|
|
|
|
rs_map: dict[str, str] = {}
|
|
for rs in apps_v1.list_replica_set_for_all_namespaces().items:
|
|
ns = rs.metadata.namespace
|
|
for ref in rs.metadata.owner_references or []:
|
|
if ref.kind == "Deployment":
|
|
rs_map[f"{ns}/{rs.metadata.name}"] = ref.name
|
|
|
|
job_map: dict[str, str] = {}
|
|
for job in batch_v1.list_job_for_all_namespaces().items:
|
|
ns = job.metadata.namespace
|
|
for ref in job.metadata.owner_references or []:
|
|
if ref.kind == "CronJob":
|
|
job_map[f"{ns}/{job.metadata.name}"] = ref.name
|
|
|
|
return rs_map, job_map
|
|
|
|
|
|
def _resolve_owner(pod, rs_map: dict, job_map: dict) -> tuple[str, str]:
|
|
ns = pod.metadata.namespace
|
|
for ref in pod.metadata.owner_references or []:
|
|
if ref.kind == "ReplicaSet":
|
|
deploy = rs_map.get(f"{ns}/{ref.name}")
|
|
return (deploy, "Deployment") if deploy else (ref.name, "ReplicaSet")
|
|
if ref.kind == "StatefulSet":
|
|
return ref.name, "StatefulSet"
|
|
if ref.kind == "DaemonSet":
|
|
return ref.name, "DaemonSet"
|
|
if ref.kind == "Job":
|
|
cj = job_map.get(f"{ns}/{ref.name}")
|
|
return (cj, "CronJob") if cj else (ref.name, "Job")
|
|
return pod.metadata.name, "Pod"
|
|
|
|
|
|
def _build_inventory() -> list[dict]:
|
|
"""Query K8s and build an app-grouped inventory. Version statuses start as 'pending'."""
|
|
_load_k8s()
|
|
v1 = client.CoreV1Api()
|
|
pods = v1.list_pod_for_all_namespaces(watch=False)
|
|
rs_map, job_map = _build_owner_maps()
|
|
|
|
apps: dict[tuple, dict] = {}
|
|
for pod in pods.items:
|
|
ns = pod.metadata.namespace
|
|
app_name, kind = _resolve_owner(pod, rs_map, job_map)
|
|
key = (ns, app_name)
|
|
if key not in apps:
|
|
apps[key] = {"namespace": ns, "app": app_name, "kind": kind, "pod_count": 0, "containers": {}}
|
|
apps[key]["pod_count"] += 1
|
|
for c in pod.spec.containers or []:
|
|
if not c.image or c.name in apps[key]["containers"]:
|
|
continue
|
|
img = parse_image_ref(c.image)
|
|
apps[key]["containers"][c.name] = {
|
|
"name": c.name,
|
|
"image": c.image,
|
|
"image_short": img.display_name,
|
|
"tag": img.tag,
|
|
"registry": img.registry,
|
|
"is_local": img.is_local,
|
|
"is_floating": img.is_floating,
|
|
"skip_check": img.skip_check,
|
|
"latest_tag": None,
|
|
"status": "pending",
|
|
}
|
|
|
|
return sorted(
|
|
[{**a, "containers": list(a["containers"].values())} for a in apps.values()],
|
|
key=lambda x: (x["namespace"], x["app"]),
|
|
)
|
|
|
|
|
|
# ── Version checking ──────────────────────────────────────────────────────────
|
|
|
|
def _check_container(container: dict) -> None:
|
|
if container["is_local"]:
|
|
container["status"] = "local"
|
|
return
|
|
if container["skip_check"]:
|
|
container["status"] = "skipped"
|
|
return
|
|
if container["is_floating"]:
|
|
container["status"] = "floating"
|
|
return
|
|
img = parse_image_ref(container["image"])
|
|
latest = get_latest_version(img)
|
|
container["latest_tag"] = latest
|
|
if latest is None:
|
|
container["status"] = "unknown"
|
|
return
|
|
try:
|
|
outdated = Version(latest.lstrip("v")) > Version(container["tag"].lstrip("v"))
|
|
except InvalidVersion:
|
|
outdated = container["tag"] != latest
|
|
container["status"] = "outdated" if outdated else "current"
|
|
|
|
|
|
def _run_checks(inventory: list[dict]) -> None:
|
|
if _cache["is_checking"]:
|
|
return
|
|
_cache["is_checking"] = True
|
|
try:
|
|
containers = [c for a in inventory for c in a["containers"]]
|
|
futures = {_pool.submit(_check_container, c): c for c in containers}
|
|
for f in as_completed(futures):
|
|
try:
|
|
f.result()
|
|
except Exception as e:
|
|
logger.warning("Version check error: %s", e)
|
|
_cache["last_updated"] = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
|
logger.info(
|
|
"Version check complete: %d outdated, %d current, %d floating",
|
|
sum(1 for a in inventory for c in a["containers"] if c["status"] == "outdated"),
|
|
sum(1 for a in inventory for c in a["containers"] if c["status"] == "current"),
|
|
sum(1 for a in inventory for c in a["containers"] if c["status"] == "floating"),
|
|
)
|
|
finally:
|
|
_cache["is_checking"] = False
|
|
|
|
|
|
def _get_inventory(background_tasks: BackgroundTasks) -> list[dict]:
|
|
"""Return cached inventory, building it and kicking off version checks if empty."""
|
|
if _cache["inventory"] is None:
|
|
inventory = _build_inventory()
|
|
_cache["inventory"] = inventory
|
|
background_tasks.add_task(_run_checks, inventory)
|
|
return _cache["inventory"]
|
|
|
|
|
|
# ── Upgrade job helpers ───────────────────────────────────────────────────────
|
|
|
|
def _list_upgrade_jobs() -> list[dict]:
|
|
try:
|
|
_load_k8s()
|
|
batch_v1 = client.BatchV1Api()
|
|
jobs = batch_v1.list_namespaced_job(
|
|
"k8s-version-tracker", label_selector="app=upgrade-requester"
|
|
)
|
|
result = []
|
|
for j in sorted(
|
|
jobs.items,
|
|
key=lambda x: x.metadata.creation_timestamp or datetime.min,
|
|
reverse=True,
|
|
)[:5]:
|
|
s = j.status
|
|
status = "running" if (s.active or 0) else ("succeeded" if (s.succeeded or 0) else "failed")
|
|
result.append({
|
|
"name": j.metadata.name,
|
|
"created": j.metadata.creation_timestamp.isoformat() if j.metadata.creation_timestamp else "",
|
|
"status": status,
|
|
"active": s.active or 0,
|
|
"succeeded": s.succeeded or 0,
|
|
"failed": s.failed or 0,
|
|
})
|
|
return result
|
|
except Exception as e:
|
|
logger.error("Failed to list upgrade jobs: %s", e)
|
|
return []
|
|
|
|
|
|
# ── Stats helper ─────────────────────────────────────────────────────────────
|
|
|
|
def _stats(inventory: list[dict]) -> dict:
|
|
all_c = [c for a in inventory for c in a["containers"]]
|
|
return {
|
|
"total": len(all_c),
|
|
"outdated": sum(1 for c in all_c if c["status"] == "outdated"),
|
|
"current": sum(1 for c in all_c if c["status"] == "current"),
|
|
"floating": sum(1 for c in all_c if c["status"] == "floating"),
|
|
"local": sum(1 for c in all_c if c["status"] in ("local", "skipped")),
|
|
"pending": sum(1 for c in all_c if c["status"] in ("pending", "unknown")),
|
|
}
|
|
|
|
|
|
# ── Routes ────────────────────────────────────────────────────────────────────
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def images_page(request: Request, background_tasks: BackgroundTasks):
|
|
inventory = _get_inventory(background_tasks)
|
|
|
|
ns_map: dict[str, list] = {}
|
|
for app_data in inventory:
|
|
ns_map.setdefault(app_data["namespace"], []).append(app_data)
|
|
|
|
return templates.TemplateResponse("images.html", {
|
|
"request": request,
|
|
"namespaces": dict(sorted(ns_map.items())),
|
|
"stats": _stats(inventory),
|
|
"last_updated": _cache["last_updated"],
|
|
"is_checking": _cache["is_checking"],
|
|
})
|
|
|
|
|
|
@app.get("/upgrade", response_class=HTMLResponse)
|
|
async def upgrade_page(request: Request):
|
|
inventory = _cache["inventory"] or []
|
|
outdated_apps = [
|
|
{**a, "containers": [c for c in a["containers"] if c["status"] == "outdated"]}
|
|
for a in inventory
|
|
if any(c["status"] == "outdated" for c in a["containers"])
|
|
]
|
|
return templates.TemplateResponse("upgrade.html", {
|
|
"request": request,
|
|
"outdated_apps": outdated_apps,
|
|
"jobs": _list_upgrade_jobs(),
|
|
"last_updated": _cache["last_updated"],
|
|
"is_checking": _cache["is_checking"],
|
|
})
|
|
|
|
|
|
@app.post("/api/refresh")
|
|
async def api_refresh(background_tasks: BackgroundTasks):
|
|
if _cache["is_checking"]:
|
|
return {"status": "already_running"}
|
|
inventory = _build_inventory()
|
|
_cache["inventory"] = inventory
|
|
background_tasks.add_task(_run_checks, inventory)
|
|
return {"status": "started"}
|
|
|
|
|
|
@app.get("/api/images")
|
|
async def api_images():
|
|
return {"inventory": _cache["inventory"] or [], "last_updated": _cache["last_updated"]}
|
|
|
|
|
|
@app.post("/api/upgrade/trigger")
|
|
async def api_trigger_upgrade():
|
|
try:
|
|
_load_k8s()
|
|
batch_v1 = client.BatchV1Api()
|
|
cj = batch_v1.read_namespaced_cron_job("upgrade-requester", "k8s-version-tracker")
|
|
stamp = datetime.now(timezone.utc).strftime("%m%d%H%M")
|
|
job_name = f"upgrade-manual-{stamp}"
|
|
job = client.V1Job(
|
|
api_version="batch/v1",
|
|
kind="Job",
|
|
metadata=client.V1ObjectMeta(
|
|
name=job_name,
|
|
namespace="k8s-version-tracker",
|
|
labels={"app": "upgrade-requester", "triggered-by": "dashboard"},
|
|
),
|
|
spec=cj.spec.job_template.spec,
|
|
)
|
|
batch_v1.create_namespaced_job("k8s-version-tracker", job)
|
|
logger.info("Triggered upgrade job: %s", job_name)
|
|
return {"status": "created", "job_name": job_name}
|
|
except Exception as e:
|
|
logger.error("Trigger failed: %s", e)
|
|
return JSONResponse(status_code=500, content={"error": str(e)})
|
|
|
|
|
|
@app.get("/api/upgrade/jobs")
|
|
async def api_upgrade_jobs():
|
|
return {"jobs": _list_upgrade_jobs()}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8080, log_level="info")
|