mirror of
https://gitea.fenix-dev.com/fenix-gitea-admin/iac-ansible-private.git
synced 2026-05-14 08:35:22 +00:00
230 lines
9.1 KiB
YAML
230 lines
9.1 KiB
YAML
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 == "<none>":
|
|
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() |