fix: add request_upgrades.py to image, add web dashboard
- Dockerfile: add COPY for request_upgrades.py (was missing — root cause of Monday upgrade CronJob failing with 'No such file'), add git (needed for repo cloning in upgrade workflow), copy web.py + templates/ - requirements.txt: add fastapi, uvicorn, jinja2 for web dashboard - web.py: FastAPI dashboard with image inventory and upgrade trigger pages - templates/: base layout, images page (namespace→app→container grouped with live version status), upgrade page (outdated list + trigger button + job history) - kubernetes/web-deployment.yaml: Deployment + LoadBalancer at 192.168.87.13 - kubernetes/rbac.yaml: add version-tracker-web ServiceAccount, ClusterRole (read pods/deployments/replicasets/statefulsets/daemonsets/jobs/cronjobs), and namespace Role to create upgrade Jobs - kubernetes/upgrade-cronjob.yaml: fix YAML indentation bug on GITEA_USERNAME and GITEA_COMMITTER_NAME env vars
This commit is contained in:
claude-code
@@ -0,0 +1,307 @@
|
||||
#!/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
|
||||
import os
|
||||
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
|
||||
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")
|
||||
Reference in New Issue
Block a user