"""Zustandsabgleich: was läuft wirklich? (Vertrag: GET /health → core.version, bundle.source/version/stale)""" from __future__ import annotations import time from concurrent.futures import ThreadPoolExecutor from sqlalchemy import select from app.db import session from app.models import Tenant from app.services import contract, deploy _CACHE: dict = {"ts": 0.0, "data": None} TTL = 30.0 def _http_get(url: str): import httpx r = httpx.get(url, timeout=6, follow_redirects=True) return r.status_code, (r.json() if r.headers.get("content-type", "").startswith("application/json") else {}) def probe(host: str, expect_tag: str, expect_bundle: int | None, get=_http_get) -> dict: out = {"host": host, "reachable": False, "core": None, "bundle": None, "drift": [], "error": None} try: status, body = get(f"https://{host}/health") except Exception as exc: # noqa: BLE001 out["error"] = f"{type(exc).__name__}: {str(exc)[:120]}" out["drift"].append("nicht erreichbar") return out out["reachable"] = status == 200 if status != 200: out["drift"].append(f"HTTP {status}") core, bundle = body.get("core"), body.get("bundle") out["core"], out["bundle"] = core, bundle if core is None: out["drift"].append("Core ohne Vertrag (alte Version)") elif expect_tag and deploy._hex_tag(expect_tag) and not contract.same_sha(core.get("version", ""), expect_tag): out["drift"].append(f"Core {core.get('version', '')[:10]} statt {expect_tag[:10]}") if expect_bundle is not None: if not bundle: out["drift"].append("meldet keine Bundle-Quelle") elif bundle.get("source") != "admin" or bundle.get("version") != expect_bundle or bundle.get("stale"): out["drift"].append(f"Bundle {bundle.get('source')} v{bundle.get('version')}{' (veraltet)' if bundle.get('stale') else ''} statt admin v{expect_bundle}") elif bundle and bundle.get("source") not in ("git", None): out["drift"].append(f"Bundle-Quelle {bundle.get('source')}, erwartet git") return out def probe_all(get=_http_get, refresh: bool = False, workers: int = 6) -> dict: if not refresh and _CACHE["data"] is not None and time.time() - _CACHE["ts"] < TTL: return _CACHE["data"] with session() as db: items = [] for t in db.scalars(select(Tenant).where(Tenant.status == "active").order_by(Tenant.id)): items.append((t.id, deploy.host_for(db, t), deploy.effective_tag(db, t), t.current_version if t.source == "admin" else None)) with ThreadPoolExecutor(max_workers=workers) as ex: res = list(ex.map(lambda it: (it[0], probe(it[1], it[2], it[3], get)), items)) data = {tid: r for tid, r in res} _CACHE.update(ts=time.time(), data=data) return data