apiVersion: v1 kind: ConfigMap metadata: name: uptime-kuma-sync-script namespace: monitoring data: sync.py: | import subprocess import sys import time import inspect import types subprocess.run([sys.executable, "-m", "pip", "install", "uptime-kuma-api-v2", "--quiet"], check=True) subprocess.run([ "bash", "-c", "curl -LO https://dl.k8s.io/release/$(curl -Ls https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl && " "chmod +x kubectl && mv kubectl /usr/local/bin/kubectl" ], check=True) from uptime_kuma_api import UptimeKumaApi, MonitorType import os # ============================================================ # CONFIGURAÇÃO # ============================================================ NOTIFICATION_IDS = [1] FIXED_TAGS = ["k8s", "IAC"] STATUS_PAGE_SLUG = "fenix" STATUS_PAGE_TITLE = "Fenix IAC" # ============================================================ UPTIME_KUMA_URL = os.environ["UPTIME_KUMA_URL"] USERNAME = os.environ["USERNAME"] PASSWORD = os.environ["PASSWORD"] print("==> A autenticar no Uptime Kuma...") api = UptimeKumaApi(UPTIME_KUMA_URL) api.login(USERNAME, PASSWORD) print("==> Autenticado com sucesso") # ── Monkey-patch _build_status_page_data ───────────────────── original_build = api._build_status_page_data.__func__ def patched_build(self, **kwargs): result = original_build(self, **kwargs) print(f" [DEBUG] type(result): {type(result)}") print(f" [DEBUG] result: {result}") slug, data, icon, public_group_list = result data.pop("googleAnalyticsId", None) return (slug, data, icon, public_group_list) api._build_status_page_data = types.MethodType(patched_build, api) print("==> Patch aplicado ao _build_status_page_data") # ── Tags ───────────────────────────────────────────────────── print("==> A sincronizar tags...") existing_tags = {t["name"]: t["id"] for t in api.get_tags()} def ensure_tag(name, color="#0099ff"): if name not in existing_tags: print(f" [TAG] A criar tag '{name}'...") result = api.add_tag(name=name, color=color) existing_tags[name] = result["id"] return existing_tags[name] ensure_tag("k8s", color="#326CE5") ensure_tag("IAC", color="#7B42BC") # ── Monitores existentes ────────────────────────────────────── print("==> A obter monitores existentes...") existing_monitors = api.get_monitors() existing_names = {m["name"] for m in existing_monitors} print(f" {len(existing_names)} monitores existentes") # ── Garantir grupo fenix ────────────────────────────────────── print("==> A verificar grupo 'fenix'...") fenix_group_id = None for m in existing_monitors: if m["name"] == "fenix" and m["type"] == "group": fenix_group_id = m["id"] print(f" [OK] Grupo 'fenix' já existe (ID: {fenix_group_id})") break if fenix_group_id is None: print(" [CRIAR] A criar grupo 'fenix'...") group = api.add_monitor(type=MonitorType.GROUP, name="fenix") fenix_group_id = group["monitorID"] print(f" [OK] Grupo 'fenix' criado (ID: {fenix_group_id})") # ── Services do cluster ─────────────────────────────────────── print("==> A listar Services do cluster...") result = subprocess.run( [ "kubectl", "get", "svc", "-A", "--no-headers", "-o", "custom-columns=NAMESPACE:.metadata.namespace,NAME:.metadata.name,PORT:.spec.ports[0].port,TYPE:.spec.type" ], capture_output=True, text=True ) services = [] for line in result.stdout.strip().split("\n"): parts = line.split() if len(parts) < 3: continue namespace, name, port = parts[0], parts[1], parts[2] if name == "kubernetes" or port == "": continue services.append((namespace, name, port)) print(f" {len(services)} services encontrados") # ── Criar monitores ─────────────────────────────────────────── created = 0 skipped = 0 for namespace, name, port in services: monitor_name = f"{namespace}/{name}" hostname = f"{name}.{namespace}.svc.cluster.local" if monitor_name in existing_names: print(f" [SKIP] {monitor_name}") skipped += 1 continue print(f" [CRIAR] {monitor_name} ({hostname}:{port})") try: ensure_tag(namespace, color="#10B981") monitor = api.add_monitor( type=MonitorType.PORT, name=monitor_name, hostname=hostname, port=int(port), interval=60, retryInterval=60, maxretries=3, parent=fenix_group_id, notificationIDList={str(nid): True for nid in NOTIFICATION_IDS}, ) monitor_id = monitor["monitorID"] api.add_monitor_tag(tag_id=existing_tags["k8s"], monitor_id=monitor_id) api.add_monitor_tag(tag_id=existing_tags["IAC"], monitor_id=monitor_id) api.add_monitor_tag(tag_id=existing_tags[namespace], monitor_id=monitor_id) print(f" [OK] {monitor_name} criado com tags e notificações") created += 1 except Exception as e: print(f" [ERRO] {monitor_name}: {e}") # ── Refrescar lista de monitores após criação ───────────────── existing_monitors = api.get_monitors() # ── Status Page ─────────────────────────────────────────────── print("==> A atualizar status page...") try: existing_pages = api.get_status_pages() page_exists = any(p["slug"] == STATUS_PAGE_SLUG for p in existing_pages) if not page_exists: print(f" [CRIAR] A criar status page '{STATUS_PAGE_SLUG}'...") api.add_status_page(STATUS_PAGE_SLUG, STATUS_PAGE_TITLE) time.sleep(5) print(f" [OK] Status page criada") current = api.get_status_page(STATUS_PAGE_SLUG) all_fenix_monitor_ids = [m["id"] for m in existing_monitors if m.get("parent") == fenix_group_id] existing_in_page = [] for group in current.get("publicGroupList", []): for mon in group.get("monitorList", []): existing_in_page.append(mon["id"]) missing_ids = [mid for mid in all_fenix_monitor_ids if mid not in existing_in_page] print(f" [DEBUG] all_fenix_monitor_ids: {all_fenix_monitor_ids}") print(f" [DEBUG] missing_ids: {missing_ids}") if not missing_ids: print(f" [SKIP] Todos os monitores já estão na status page") else: public_group_list = current.get("publicGroupList", []) if public_group_list: for mid in missing_ids: public_group_list[0]["monitorList"].append({"id": mid}) else: public_group_list = [ { "name": "Fenix IAC K8s", "weight": 1, "monitorList": [{"id": mid} for mid in all_fenix_monitor_ids], } ] print(f" [DEBUG] publicGroupList: {public_group_list}") api.save_status_page( slug=STATUS_PAGE_SLUG, id=current["id"], title=current.get("title", STATUS_PAGE_TITLE), description=current.get("description"), theme=current.get("theme", "auto"), published=current.get("published", True), showTags=current.get("showTags", True), domainNameList=current.get("domainNameList", []), customCSS=current.get("customCSS") or "", footerText=current.get("footerText"), showPoweredBy=current.get("showPoweredBy", True), showCertificateExpiry=current.get("showCertificateExpiry", False), icon=current.get("icon", "/icon.svg"), publicGroupList=public_group_list, ) print(f" [OK] Status page atualizada — {len(missing_ids)} monitores adicionados") print(f" URL: {UPTIME_KUMA_URL}/status/{STATUS_PAGE_SLUG}") except Exception as e: print(f" [ERRO] Status page: {e}") import traceback traceback.print_exc() print(f"==> Sync concluído — {created} criados, {skipped} ignorados") api.disconnect()