JiriUhlir
ed194093e6
Přehled má nové karty: Zdravé služby, Nezdravé služby, Nedostupné služby. Přehled zobrazuje sekci Služby vyžadující pozornost pro unhealthy a unreachable. Snapshot /portal/ws/operations teď obsahuje health data a stránka je aktualizuje přes existující realtime vrstvu. Seznam služeb zobrazuje aktuální health status a poslední kontrolu, včetně řazení Podle zdraví. Detail služby má sekce Zdraví služby a Historie kontrol s posledními 50 záznamy. Přidal jsem audit event service.health.view. Browser title teď používá formát CSBot Services Portal - ..., takže detail služby odpovídá požadavku. Nepoužil jsem ORM, vše je přes SQLite dotazy.
103 lines
2.6 KiB
Python
103 lines
2.6 KiB
Python
from app.db.database import get_connection
|
|
|
|
|
|
HEALTH_STATUSES = ("healthy", "unhealthy", "unreachable")
|
|
|
|
|
|
def get_latest_service_health():
|
|
con = get_connection()
|
|
|
|
try:
|
|
rows = con.execute(
|
|
"""
|
|
SELECT
|
|
h.service_id,
|
|
COALESCE(a.name, h.service_id) AS service_name,
|
|
h.status,
|
|
h.http_status,
|
|
h.response_time_ms,
|
|
h.error_text,
|
|
h.checked_at
|
|
FROM service_health h
|
|
LEFT JOIN apps a ON a.id = h.service_id
|
|
INNER JOIN (
|
|
SELECT service_id, MAX(id) AS max_id
|
|
FROM service_health
|
|
GROUP BY service_id
|
|
) latest ON latest.max_id = h.id
|
|
"""
|
|
).fetchall()
|
|
except Exception:
|
|
rows = []
|
|
|
|
con.close()
|
|
return {row["service_id"]: dict(row) for row in rows}
|
|
|
|
|
|
def get_service_health(service_id: str):
|
|
con = get_connection()
|
|
|
|
try:
|
|
row = con.execute(
|
|
"""
|
|
SELECT
|
|
h.service_id,
|
|
COALESCE(a.name, h.service_id) AS service_name,
|
|
h.status,
|
|
h.http_status,
|
|
h.response_time_ms,
|
|
h.error_text,
|
|
h.checked_at
|
|
FROM service_health h
|
|
LEFT JOIN apps a ON a.id = h.service_id
|
|
WHERE h.service_id = ?
|
|
ORDER BY h.id DESC
|
|
LIMIT 1
|
|
""",
|
|
(service_id,),
|
|
).fetchone()
|
|
except Exception:
|
|
row = None
|
|
|
|
con.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def get_service_health_history(service_id: str, limit: int = 50):
|
|
con = get_connection()
|
|
|
|
try:
|
|
rows = con.execute(
|
|
"""
|
|
SELECT id, service_id, status, http_status, response_time_ms, error_text, checked_at
|
|
FROM service_health
|
|
WHERE service_id = ?
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
""",
|
|
(service_id, limit),
|
|
).fetchall()
|
|
except Exception:
|
|
rows = []
|
|
|
|
con.close()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def get_health_summary():
|
|
latest = get_latest_service_health()
|
|
summary = {"healthy": 0, "unhealthy": 0, "unreachable": 0}
|
|
for item in latest.values():
|
|
status = item.get("status")
|
|
if status in summary:
|
|
summary[status] += 1
|
|
return summary
|
|
|
|
|
|
def get_problem_service_health():
|
|
latest = get_latest_service_health()
|
|
return [
|
|
item for item in latest.values()
|
|
if item.get("status") in {"unhealthy", "unreachable"}
|
|
]
|