Micha/homelab-infra
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

apps/dashboard/backend/app/clients/uptime_kuma_client.py aktualisiert

Browse Source
This commit is contained in:
Micha
2026-04-06 12:51:39 +00:00
parent 043b5697fd
commit 9f7cc82554
1 changed files with 19 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/dashboard/backend/app/clients/uptime_kuma_client.py
+19 -29
View File
@@ -9,17 +9,19 @@ from app.clients.base import BaseHTTPClient
from app.config import Settings
from app.models.sources import UptimeKumaMonitor, UptimeKumaSnapshot
METRIC_LINE_RE = re.compile(r'^(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*){(?P<labels>[^}]*)}s+(?P<value>.+)$')
LABEL_RE = re.compile(r'(w+)="((?:[^"\\]|\\.)*)"')
METRIC_LINE_RE = re.compile(r'^(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*){(?P<labels>[^}]*)}\s+(?P<value>.+)$')
LABEL_RE = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"')
logger = logging.getLogger(__name__)
class UptimeKumaClient(BaseHTTPClient):
"""
Reads Uptime Kuma monitor status from the documented /metrics endpoint.
This avoids coupling the backend to Socket.IO login flows, but still relies
on Kuma's internal metrics surface, which may change across releases.
Reads Uptime Kuma monitor status from the /metrics endpoint.
Auth: once an API key exists in Uptime Kuma, username/password Basic Auth
is permanently disabled. The correct format is HTTP Basic Auth with an
empty username and the API key as the password: auth=("", api_key).
"""
def __init__(self, settings: Settings) -> None:
@@ -33,12 +35,14 @@ class UptimeKumaClient(BaseHTTPClient):
raw_metrics: str | None = None
# Primary: API key as Basic Auth password (empty username)
if self.settings.uptime_kuma_api_key:
raw_metrics = await self._request_metrics_with_mode(
"api-key",
headers={"Authorization": f"Bearer {self.settings.uptime_kuma_api_key}"},
auth=("", self.settings.uptime_kuma_api_key),
)
# Fallback: regular username/password (only works if no API keys exist)
if raw_metrics is None and self.settings.uptime_kuma_username and self.settings.uptime_kuma_password:
raw_metrics = await self._request_metrics_with_mode(
"basic-user",
@@ -57,12 +61,10 @@ class UptimeKumaClient(BaseHTTPClient):
return snapshot
logger.info("uptime kuma raw metrics first 40 lines: %s", raw_metrics.splitlines()[:40])
monitors = self._parse_metrics(raw_metrics)
up = sum(1 for monitor in monitors if monitor.status == "online")
down = sum(1 for monitor in monitors if monitor.status == "offline")
paused = sum(1 for monitor in monitors if monitor.status == "degraded")
normalized = UptimeKumaSnapshot(
source_status="online",
monitors_up=up,
@@ -78,34 +80,26 @@ class UptimeKumaClient(BaseHTTPClient):
self,
mode: str,
*,
headers: dict[str, str] | None = None,
auth: tuple[str, str] | None = None,
) -> str | None:
if not self.base_url:
return None
url = f"{self.base_url}/metrics"
try:
async with httpx.AsyncClient(
timeout=self.settings.request_timeout_seconds,
trust_env=False,
) as client:
response = await client.request(
"GET",
url,
headers=headers,
auth=auth,
response = await client.request("GET", url, auth=auth)
if response.status_code == 200 and response.text:
logger.info("uptime kuma metrics auth succeeded via %s", mode)
return response.text
if response.status_code in (401, 403):
logger.warning(
"uptime kuma metrics auth failed via %s with status %s",
mode,
response.status_code,
)
if response.status_code == 200 and response.text:
logger.info("uptime kuma metrics auth succeeded via %s", mode)
return response.text
if response.status_code in (401, 403):
logger.warning(
"uptime kuma metrics auth failed via %s with status %s",
mode,
response.status_code,
)
return None
except httpx.TimeoutException:
logger.warning("uptime kuma metrics request timed out via %s", mode)
@@ -115,25 +109,21 @@ class UptimeKumaClient(BaseHTTPClient):
def _parse_metrics(self, payload: str) -> list[UptimeKumaMonitor]:
status_by_id: dict[str, UptimeKumaMonitor] = {}
for line in payload.splitlines():
parsed = self._parse_metric_line(line)
if parsed is None:
continue
metric_name, labels, raw_value = parsed
monitor_id = labels.get("monitor_id") or labels.get("id") or labels.get("monitor") or labels.get("monitor_name")
monitor_name = labels.get("monitor_name") or labels.get("name")
if not monitor_id or not monitor_name:
continue
if monitor_id not in status_by_id:
status_by_id[monitor_id] = UptimeKumaMonitor(
id=self._as_int(monitor_id),
name=monitor_name,
)
monitor = status_by_id[monitor_id]
if metric_name == "monitor_status":
status_code = self._as_int_from_float(raw_value)
if status_code == 1: