from __future__ import annotations import json import time from typing import Any from urllib.error import URLError from urllib.parse import urlencode from urllib.request import urlopen from flask import jsonify from .. import settings _LAB_STATUS_CACHE: dict[str, Any] = {"ts": 0.0, "value": None} def _status_response(payload: dict[str, Any]) -> Any: """Return lab status JSON with browser and proxy caching disabled.""" response = jsonify(payload) response.headers["Cache-Control"] = "no-store, max-age=0" response.headers["Pragma"] = "no-cache" return response def _vm_query(expr: str) -> float | None: """Run one instant VictoriaMetrics query and return the largest value.""" url = f"{settings.VM_BASE_URL}/api/v1/query?{urlencode({'query': expr})}" with urlopen(url, timeout=settings.VM_QUERY_TIMEOUT_SEC) as resp: payload = json.loads(resp.read().decode("utf-8")) if payload.get("status") != "success": return None result = (payload.get("data") or {}).get("result") or [] if not result: return None values: list[float] = [] for item in result: try: values.append(float(item["value"][1])) except (KeyError, IndexError, TypeError, ValueError): continue if not values: return None return max(values) def _vm_query_vector(expr: str) -> list[dict[str, Any]]: """Run one instant VictoriaMetrics query and return result samples.""" url = f"{settings.VM_BASE_URL}/api/v1/query?{urlencode({'query': expr})}" with urlopen(url, timeout=settings.VM_QUERY_TIMEOUT_SEC) as resp: payload = json.loads(resp.read().decode("utf-8")) if payload.get("status") != "success": return [] result = (payload.get("data") or {}).get("result") or [] samples: list[dict[str, Any]] = [] for item in result: try: value = float(item["value"][1]) except (KeyError, IndexError, TypeError, ValueError): continue metric = item.get("metric") if isinstance(item.get("metric"), dict) else {} samples.append({"metric": metric, "value": value}) return samples def _http_ok(url: str, expect_substring: str | None = None) -> bool: """Return whether a URL responds successfully and optionally contains text.""" try: with urlopen(url, timeout=settings.HTTP_CHECK_TIMEOUT_SEC) as resp: if getattr(resp, "status", 200) != 200: return False if expect_substring: chunk = resp.read(4096).decode("utf-8", errors="ignore") return expect_substring in chunk return True except (URLError, TimeoutError, ValueError): return False _PUBLIC_SERVICE_ALIASES = [ (("cloud.bstein.dev", "nextcloud", "cloud"), "Nextcloud", "https://cloud.bstein.dev"), (("live.bstein.dev", "element", "matrix", "synapse", "livekit"), "Element / Matrix", "https://live.bstein.dev"), (("stream.bstein.dev", "jellyfin", "stream"), "Jellyfin", "https://stream.bstein.dev"), (("notes.bstein.dev", "outline", "notes"), "Outline", "https://notes.bstein.dev"), (("tasks.bstein.dev", "planka", "tasks"), "Planka", "https://tasks.bstein.dev"), (("vault.bstein.dev", "vaultwarden"), "VaultWarden", "https://vault.bstein.dev"), (("scm.bstein.dev", "gitea"), "Gitea", "https://scm.bstein.dev"), (("ci.bstein.dev", "jenkins"), "Jenkins", "https://ci.bstein.dev"), (("registry.bstein.dev", "harbor", "registry"), "Harbor", "https://registry.bstein.dev"), (("metrics.bstein.dev", "grafana", "metrics"), "Grafana", "https://metrics.bstein.dev"), (("chat.ai.bstein.dev", "ai-chat", "atlasbot", "ollama"), "AI Chat", "https://bstein.dev/ai/chat"), (("health.bstein.dev", "wger", "health"), "Wger", "https://health.bstein.dev"), (("budget.bstein.dev", "actual", "budget"), "Actual Budget", "https://budget.bstein.dev"), (("money.bstein.dev", "firefly", "money"), "Firefly III", "https://money.bstein.dev"), (("mail.bstein.dev", "mailu", "mail"), "Mailu", "https://mail.bstein.dev"), ] def _public_active_service() -> dict[str, Any]: """Return the busiest mapped public service without leaking raw metrics.""" if not settings.LAB_ACTIVE_SERVICE_QUERY: return {"known": False} samples = _vm_query_vector(settings.LAB_ACTIVE_SERVICE_QUERY) for sample in sorted(samples, key=lambda item: item["value"], reverse=True): labels = sample.get("metric") or {} haystack = " ".join(str(value).lower() for value in labels.values()) for aliases, label, url in _PUBLIC_SERVICE_ALIASES: if any(alias in haystack for alias in aliases): return { "known": True, "label": label, "url": url, "window": settings.LAB_ACTIVE_SERVICE_WINDOW, } return {"known": False} def _dedicated_host_status(label: str, health_url: str, up_query: str) -> dict[str, Any]: """Check one dedicated host and return only public-safe status fields.""" if health_url: if _http_ok(health_url): return {"label": label, "known": True, "up": True, "source": "health-check"} return {"label": label, "known": True, "up": False, "source": "health-check"} if up_query: value = _vm_query(up_query) if value is not None: return {"label": label, "known": True, "up": value > 0, "source": "victoria-metrics"} return {"label": label, "known": False, "up": False, "source": "unknown"} def _dedicated_hosts_status() -> dict[str, Any]: """Return aggregate availability for the non-cluster DB and jumphost.""" hosts = [ _dedicated_host_status( "Database node", settings.ATLAS_DB_HOST_HEALTH_URL, settings.ATLAS_DB_HOST_UP_QUERY, ), _dedicated_host_status( "Jumphost", settings.ATLAS_JUMPHOST_HEALTH_URL, settings.ATLAS_JUMPHOST_UP_QUERY, ), ] known_hosts = [host for host in hosts if host["known"]] up_count = sum(1 for host in known_hosts if host["up"]) return { "known": bool(known_hosts), "up": bool(known_hosts) and up_count == len(known_hosts), "up_count": up_count, "total": len(hosts), "known_count": len(known_hosts), "hosts": hosts, } def register(app) -> None: """Register the lightweight lab connectivity status endpoint.""" @app.route("/api/lab/status") def lab_status() -> Any: """Return cached public-safe lab health hints for the home page.""" now = time.time() cached = _LAB_STATUS_CACHE.get("value") if cached and (now - float(_LAB_STATUS_CACHE.get("ts", 0.0)) < settings.LAB_STATUS_CACHE_SEC): return _status_response(cached) t_total = time.perf_counter() timings_ms: dict[str, int] = {} connected = False atlas_up = False atlas_known = False atlas_source = "unknown" dedicated_hosts = {"known": False, "up": False, "up_count": 0, "total": 2, "known_count": 0, "hosts": []} active_service = {"known": False} # Atlas try: t_probe = time.perf_counter() atlas_grafana_ok = _http_ok(settings.GRAFANA_HEALTH_URL, expect_substring="ok") timings_ms["grafana"] = int((time.perf_counter() - t_probe) * 1000) if atlas_grafana_ok: connected = True atlas_up = True atlas_known = True atlas_source = "grafana" except Exception: pass if not atlas_known: try: t_probe = time.perf_counter() value = _vm_query("up") timings_ms["victoria_metrics"] = int((time.perf_counter() - t_probe) * 1000) if value is not None: connected = True atlas_known = True atlas_up = value > 0 atlas_source = "victoria-metrics" except Exception: pass try: t_probe = time.perf_counter() dedicated_hosts = _dedicated_hosts_status() timings_ms["dedicated_hosts"] = int((time.perf_counter() - t_probe) * 1000) if dedicated_hosts["known"]: connected = True except Exception: pass try: t_probe = time.perf_counter() active_service = _public_active_service() timings_ms["active_service"] = int((time.perf_counter() - t_probe) * 1000) if active_service.get("known"): connected = True except Exception: pass timings_ms["total"] = int((time.perf_counter() - t_total) * 1000) payload = { "connected": connected, "atlas": {"up": atlas_up, "known": atlas_known, "source": atlas_source}, "dedicated_hosts": dedicated_hosts, "active_service": active_service, "oceanus": {"up": atlas_up, "known": atlas_known, "source": "atlas-member"}, "checked_at": int(now), "timings_ms": timings_ms, } _LAB_STATUS_CACHE["ts"] = now _LAB_STATUS_CACHE["value"] = payload return _status_response(payload)