Hotovo. Přidal jsem správu Alertingu do appfactory-portal bez ORM:

Nové SQL migrace pro alert_rules a alert_events v migrations.py (line 165).
Nová ruční DB vrstva v alerting.py (line 7).
Nové stránky a akce v routes/alerting.py (line 288):Alert pravidla, detail, vytvoření, úprava, smazání
povolit/zakázat pravidlo
Alert eventy a detail eventu s payload_json
odkazy z job_id na detail úlohy
detail pravidla s posledními eventy
admin-only editor .sh alert skriptů v /opt/appfactory/workspace/appfactory-tools/alerts
audit eventy podle zadání

Menu Provoz → Alerting v layout.py (line 23).
Router je zaregistrovaný v main.py (line 8).
This commit is contained in:
JiriUhlir
2026-06-08 06:57:12 +02:00
parent 64f8fe806d
commit 382825085f
5 changed files with 941 additions and 1 deletions
+653
View File
@@ -0,0 +1,653 @@
import html
import os
from pathlib import Path
from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from app.auth import require_user
from app.db.alerting import (
create_alert_rule,
delete_alert_rule,
get_alert_event,
get_alert_events,
get_alert_rule,
get_alert_rules,
get_rule_alert_events,
set_alert_rule_enabled,
update_alert_rule,
)
from app.db.audit import log_audit_event
from app.routes.jobs import pretty_json
from app.templates.layout import page
router = APIRouter()
ALERTS_DIR = Path("/opt/appfactory/workspace/appfactory-tools/alerts")
MAX_SCRIPT_BYTES = 100 * 1024
EVENT_TYPES = ("incident.opened", "incident.resolved")
DEFAULT_SCRIPT_CONTENT = """#!/usr/bin/env bash
set -euo pipefail
echo "TODO"
"""
def clean_optional(value: str | None) -> str:
return (value or "").strip()
def is_admin(user) -> bool:
return (user.get("role") or "").lower() == "admin"
def render_enabled_pill(value) -> str:
if value:
return '<span class="pill pill-success">zapnuto</span>'
return '<span class="pill pill-muted">vypnuto</span>'
def render_status(value: str | None) -> str:
status = (value or "").lower()
if status in {"success", "processed", "done", "ok"}:
return '<span class="pill pill-success">zpracováno</span>'
if status in {"queued", "running", "pending"}:
return '<span class="pill pill-warning">čeká</span>'
if status in {"failed", "error"}:
return '<span class="pill pill-danger">selhalo</span>'
if not status:
return '<span class="pill pill-muted">bez stavu</span>'
return f'<span class="pill pill-muted">{html.escape(value or "")}</span>'
def render_event_type_options(selected: str) -> str:
options = []
for event_type in EVENT_TYPES:
selected_attr = " selected" if selected == event_type else ""
escaped = html.escape(event_type)
options.append(f'<option value="{escaped}"{selected_attr}>{escaped}</option>')
return "".join(options)
def validate_event_type(value: str) -> str:
event_type = clean_optional(value)
if event_type not in EVENT_TYPES:
raise HTTPException(status_code=400, detail="Neplatný typ události")
return event_type
def validate_script_name(value: str) -> str:
script_name = clean_optional(value)
if not script_name:
raise HTTPException(status_code=400, detail="Název skriptu je povinný")
if not script_name.endswith(".sh"):
raise HTTPException(status_code=400, detail="Název skriptu musí končit .sh")
if "/" in script_name or "\\" in script_name or ".." in script_name:
raise HTTPException(status_code=400, detail="Název skriptu nesmí obsahovat cestu")
return script_name
def script_path(script_name: str) -> Path:
safe_name = validate_script_name(script_name)
path = (ALERTS_DIR / safe_name).resolve()
base = ALERTS_DIR.resolve()
try:
path.relative_to(base)
except ValueError:
raise HTTPException(status_code=400, detail="Neplatný název skriptu")
return path
def read_script_file(script_name: str) -> dict:
path = script_path(script_name)
if not path.exists():
return {
"content": DEFAULT_SCRIPT_CONTENT,
"error": None,
"warning": "Soubor zatím neexistuje. Administrátor ho může vytvořit z výchozího obsahu.",
}
if not path.is_file():
return {"content": "", "error": "Cesta není soubor.", "warning": None}
try:
if path.stat().st_size > MAX_SCRIPT_BYTES:
return {"content": "", "error": "Soubor je větší než 100 KB.", "warning": None}
content = path.read_text(encoding="utf-8")
except Exception:
return {"content": "", "error": "Soubor se nepodařilo načíst.", "warning": None}
warning = None
if not content.startswith("#!/usr/bin/env bash"):
warning = "První řádek by měl být #!/usr/bin/env bash."
return {"content": content, "error": None, "warning": warning}
def normalize_script_content(content: str) -> str:
value = content.replace("\r\n", "\n").replace("\r", "\n")
if not value.strip():
raise HTTPException(status_code=400, detail="Obsah skriptu nesmí být prázdný")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je větší než 100 KB")
if not value.startswith("#!/usr/bin/env bash"):
value = "#!/usr/bin/env bash\n" + value.lstrip("\n")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je větší než 100 KB")
return value
def save_script_file(script_name: str, content: str) -> None:
path = script_path(script_name)
value = normalize_script_content(content)
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(value, encoding="utf-8", newline="\n")
os.chmod(path, 0o755)
except Exception:
raise HTTPException(status_code=500, detail="Soubor se nepodařilo uložit")
def form_metadata(
name: str,
description: str,
event_type: str,
service_id: str,
script_name: str,
is_enabled: str | None,
) -> dict:
name_value = clean_optional(name)
if not name_value:
raise HTTPException(status_code=400, detail="Název je povinný")
return {
"name": name_value,
"description": clean_optional(description),
"event_type": validate_event_type(event_type),
"service_id": clean_optional(service_id),
"script_name": validate_script_name(script_name),
"is_enabled": bool(is_enabled),
}
def render_rule_form(rule: dict | None, action: str) -> str:
rule = rule or {}
name = html.escape(rule.get("name", "") or "")
description = html.escape(rule.get("description", "") or "")
service_id = html.escape(rule.get("service_id", "") or "")
script_name = html.escape(rule.get("script_name", "log-alert.sh") or "log-alert.sh")
event_type = rule.get("event_type", EVENT_TYPES[0]) or EVENT_TYPES[0]
enabled_checked = " checked" if rule.get("is_enabled", True) else ""
return f"""
<form method="post" action="{action}" class="metadata-form">
<label>Název</label>
<input name="name" value="{name}" required>
<label>Popis</label>
<textarea name="description" rows="3">{description}</textarea>
<label>Typ události</label>
<select name="event_type">{render_event_type_options(event_type)}</select>
<label>Služba</label>
<input name="service_id" value="{service_id}" placeholder="volitelné ID služby">
<label>Skript</label>
<input name="script_name" value="{script_name}" placeholder="log-alert.sh" required>
<label class="checkbox-label"><input type="checkbox" name="is_enabled" value="1"{enabled_checked}> Zapnuto</label>
<div class="form-actions">
<button type="submit">Uložit</button>
</div>
</form>
"""
def render_event_rows(events: list[dict], empty_colspan: int = 9, include_rule: bool = True) -> str:
rows = ""
for event in events:
event_id = html.escape(str(event.get("id", "")))
rule_id = event.get("rule_id")
rule = ""
if include_rule:
if rule_id:
rule_name = html.escape(event.get("rule_name") or f"#{rule_id}")
rule = f'<td><a href="/portal/alerting/rules/{html.escape(str(rule_id))}">{rule_name}</a></td>'
else:
rule = "<td></td>"
job_id = event.get("job_id")
job_link = ""
if job_id:
job_id_html = html.escape(str(job_id))
job_link = f'<a href="/portal/jobs/{job_id_html}">#{job_id_html}</a>'
rows += f"""
<tr>
<td><strong><a href="/portal/alerting/events/{event_id}">#{event_id}</a></strong></td>
{rule}
<td>{html.escape(event.get("event_type", "") or "")}</td>
<td>{html.escape(event.get("service_id", "") or "")}</td>
<td>{html.escape(str(event.get("incident_id") or ""))}</td>
<td>{render_status(event.get("status"))}</td>
<td>{job_link}</td>
<td>{html.escape(event.get("created_at", "") or "")}</td>
<td><a class="btn" href="/portal/alerting/events/{event_id}">Detail</a></td>
</tr>
"""
if not rows:
rows = f'<tr><td colspan="{empty_colspan}">Zatím nejsou evidované žádné alert eventy.</td></tr>'
return rows
def render_script_editor(rule: dict, user: dict) -> str:
script_name = rule.get("script_name", "") or ""
script_name_html = html.escape(script_name)
try:
script_file = read_script_file(script_name)
except HTTPException:
script_file = {"content": "", "error": "Název skriptu není bezpečný.", "warning": None}
content = html.escape(script_file.get("content", "") or "")
warning = f'<p class="alert">{html.escape(script_file["warning"])}</p>' if script_file.get("warning") else ""
error = f'<p class="alert alert-danger">{html.escape(script_file["error"])}</p>' if script_file.get("error") else ""
if not is_admin(user):
return f"""
<div class="card">
<h2>Obsah alert skriptu</h2>
<p><strong>Soubor:</strong> {script_name_html}</p>
{warning}
{error}
<p class="muted">Obsah skriptu je pouze pro čtení. Ukládat ho může jen administrátor.</p>
<textarea rows="18" readonly>{content}</textarea>
</div>
"""
return f"""
<div class="card">
<h2>Editor alert skriptu</h2>
<p><strong>Soubor:</strong> {script_name_html}</p>
{warning}
{error}
<form method="post" action="/portal/alerting/rules/{html.escape(str(rule.get("id")))}/script" class="metadata-form">
<label>Obsah souboru</label>
<textarea name="content" rows="22" spellcheck="false">{content}</textarea>
<div class="form-actions">
<button type="submit">Uložit skript</button>
</div>
</form>
</div>
"""
@router.get("/alerting", response_class=HTMLResponse)
def alerting_redirect(user=Depends(require_user)):
return RedirectResponse(url="/portal/alerting/rules", status_code=303)
@router.get("/alerting/rules", response_class=HTMLResponse)
def alert_rules_page(request: Request, user=Depends(require_user)):
rows = ""
for rule in get_alert_rules():
rule_id = html.escape(str(rule.get("id", "")))
enabled = bool(rule.get("is_enabled"))
toggle_label = "Zakázat" if enabled else "Povolit"
last_event = ""
if rule.get("last_event_id"):
event_id = html.escape(str(rule.get("last_event_id")))
last_event = f'<a href="/portal/alerting/events/{event_id}">#{event_id}</a> {render_status(rule.get("last_event_status"))}'
rows += f"""
<tr>
<td><strong><a href="/portal/alerting/rules/{rule_id}">{html.escape(rule.get("name", "") or "")}</a></strong></td>
<td>{html.escape(rule.get("event_type", "") or "")}</td>
<td>{html.escape(rule.get("service_id", "") or "")}</td>
<td>{html.escape(rule.get("script_name", "") or "")}</td>
<td>{render_enabled_pill(enabled)}</td>
<td>{last_event}</td>
<td>{html.escape(rule.get("updated_at", "") or "")}</td>
<td class="actions-cell service-actions">
<a class="btn" href="/portal/alerting/rules/{rule_id}">Detail</a>
<a class="btn btn-secondary" href="/portal/alerting/rules/{rule_id}/edit">Upravit</a>
<form method="post" action="/portal/alerting/rules/{rule_id}/toggle" class="inline-form">
<button type="submit" class="btn btn-secondary">{toggle_label}</button>
</form>
<form method="post" action="/portal/alerting/rules/{rule_id}/delete" class="inline-form" onsubmit="return confirm('Smazat alert pravidlo?');">
<button type="submit" class="danger">Smazat</button>
</form>
</td>
</tr>
"""
if not rows:
rows = '<tr><td colspan="8">Zatím nejsou evidovaná žádná alert pravidla.</td></tr>'
return page(
"Alert pravidla",
f"""
<div class="card">
<h2>Alert pravidla</h2>
<p class="muted">Správa pravidel pro spouštění alert skriptů při událostech incidentů.</p>
<p>
<a class="btn" href="/portal/alerting/rules/new">+ Nové alert pravidlo</a>
<a class="btn btn-secondary" href="/portal/alerting/events">Alert eventy</a>
</p>
</div>
<div class="card">
<h2>Pravidla</h2>
<table>
<tr>
<th>Název</th>
<th>Typ události</th>
<th>Služba</th>
<th>Skript</th>
<th>Stav</th>
<th>Poslední event</th>
<th>Upraveno</th>
<th>Akce</th>
</tr>
{rows}
</table>
</div>
""",
user=user,
)
@router.get("/alerting/rules/new", response_class=HTMLResponse)
def new_alert_rule_form(request: Request, user=Depends(require_user)):
return page(
"Nové alert pravidlo",
f"""
<div class="card">
<h2>Nové alert pravidlo</h2>
<p><a class="btn" href="/portal/alerting/rules">&larr; Zpět</a></p>
</div>
<div class="card">
{render_rule_form({"is_enabled": True, "event_type": EVENT_TYPES[0], "script_name": "log-alert.sh"}, "/portal/alerting/rules/new")}
</div>
""",
user=user,
)
@router.post("/alerting/rules/new")
def create_alert_rule_action(
name: str = Form(...),
description: str = Form(""),
event_type: str = Form(...),
service_id: str = Form(""),
script_name: str = Form(...),
is_enabled: str | None = Form(None),
user=Depends(require_user),
):
metadata = form_metadata(name, description, event_type, service_id, script_name, is_enabled)
rule_id = create_alert_rule(metadata)
log_audit_event(
user,
action="alert_rule.created",
target_type="alert_rule",
target_id=rule_id,
metadata={"event_type": metadata["event_type"], "script_name": metadata["script_name"]},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.get("/alerting/rules/{rule_id}", response_class=HTMLResponse)
def alert_rule_detail(rule_id: int, request: Request, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
enabled = bool(rule.get("is_enabled"))
toggle_label = "Zakázat" if enabled else "Povolit"
events = get_rule_alert_events(rule_id, limit=20)
return page(
rule.get("name", "") or "Alert pravidlo",
f"""
<div class="card">
<h2>{html.escape(rule.get("name", "") or "")}</h2>
<p>
<a class="btn" href="/portal/alerting/rules">&larr; Zpět na alert pravidla</a>
<a class="btn btn-secondary" href="/portal/alerting/rules/{rule_id}/edit">Upravit</a>
<a class="btn btn-secondary" href="/portal/alerting/events">Alert eventy</a>
</p>
<div class="inline-form">
<form method="post" action="/portal/alerting/rules/{rule_id}/toggle">
<button type="submit" class="btn-secondary">{toggle_label}</button>
</form>
<form method="post" action="/portal/alerting/rules/{rule_id}/delete" onsubmit="return confirm('Smazat alert pravidlo?');">
<button type="submit" class="danger">Smazat</button>
</form>
</div>
</div>
<div class="card">
<h2>Souhrn</h2>
<table>
<tr><th>Název</th><td>{html.escape(rule.get("name", "") or "")}</td></tr>
<tr><th>Popis</th><td>{html.escape(rule.get("description", "") or "")}</td></tr>
<tr><th>Typ události</th><td>{html.escape(rule.get("event_type", "") or "")}</td></tr>
<tr><th>Služba</th><td>{html.escape(rule.get("service_id", "") or "")}</td></tr>
<tr><th>Skript</th><td>{html.escape(rule.get("script_name", "") or "")}</td></tr>
<tr><th>Zapnuto</th><td>{"Ano" if enabled else "Ne"}</td></tr>
<tr><th>Vytvořeno</th><td>{html.escape(rule.get("created_at", "") or "")}</td></tr>
<tr><th>Upraveno</th><td>{html.escape(rule.get("updated_at", "") or "")}</td></tr>
</table>
</div>
<div class="card">
<h2>Poslední eventy</h2>
<table>
<tr>
<th>ID</th>
<th>Typ události</th>
<th>Služba</th>
<th>Incident</th>
<th>Stav</th>
<th>Úloha</th>
<th>Vytvořeno</th>
<th>Akce</th>
</tr>
{render_event_rows(events, empty_colspan=8, include_rule=False)}
</table>
</div>
{render_script_editor(rule, user)}
""",
user=user,
)
@router.get("/alerting/rules/{rule_id}/edit", response_class=HTMLResponse)
def edit_alert_rule_form(rule_id: int, request: Request, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
return page(
"Upravit alert pravidlo",
f"""
<div class="card">
<h2>Upravit alert pravidlo</h2>
<p><a class="btn" href="/portal/alerting/rules/{rule_id}">&larr; Zpět</a></p>
</div>
<div class="card">
{render_rule_form(rule, f"/portal/alerting/rules/{rule_id}/edit")}
</div>
""",
user=user,
)
@router.post("/alerting/rules/{rule_id}/edit")
def update_alert_rule_action(
rule_id: int,
name: str = Form(...),
description: str = Form(""),
event_type: str = Form(...),
service_id: str = Form(""),
script_name: str = Form(...),
is_enabled: str | None = Form(None),
user=Depends(require_user),
):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
metadata = form_metadata(name, description, event_type, service_id, script_name, is_enabled)
update_alert_rule(rule_id, metadata)
log_audit_event(
user,
action="alert_rule.updated",
target_type="alert_rule",
target_id=rule_id,
metadata={"event_type": metadata["event_type"], "script_name": metadata["script_name"]},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.post("/alerting/rules/{rule_id}/toggle")
def toggle_alert_rule(rule_id: int, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
enabled = not bool(rule.get("is_enabled"))
set_alert_rule_enabled(rule_id, enabled)
log_audit_event(
user,
action="alert_rule.enabled" if enabled else "alert_rule.disabled",
target_type="alert_rule",
target_id=rule_id,
metadata={"script_name": rule.get("script_name")},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.post("/alerting/rules/{rule_id}/delete")
def delete_alert_rule_action(rule_id: int, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
if not delete_alert_rule(rule_id):
raise HTTPException(status_code=409, detail="Alert pravidlo nelze smazat")
log_audit_event(
user,
action="alert_rule.deleted",
target_type="alert_rule",
target_id=rule_id,
metadata={"script_name": rule.get("script_name")},
)
return RedirectResponse(url="/portal/alerting/rules", status_code=303)
@router.post("/alerting/rules/{rule_id}/script")
def update_alert_script(rule_id: int, content: str = Form(...), user=Depends(require_user)):
if not is_admin(user):
raise HTTPException(status_code=403, detail="Alert skript může upravit pouze administrátor")
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
script_name = validate_script_name(rule.get("script_name", "") or "")
save_script_file(script_name, content)
log_audit_event(
user,
action="alert_script.updated",
target_type="alert_script",
target_id=script_name,
metadata={"alert_rule_id": rule_id, "script_name": script_name},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.get("/alerting/events", response_class=HTMLResponse)
def alert_events_page(request: Request, user=Depends(require_user)):
events = get_alert_events(limit=200)
return page(
"Alert eventy",
f"""
<div class="card">
<h2>Alert eventy</h2>
<p class="muted">Historie vyvolaných alertů a jejich zpracování.</p>
<p><a class="btn" href="/portal/alerting/rules">Alert pravidla</a></p>
</div>
<div class="card">
<h2>Eventy</h2>
<table>
<tr>
<th>ID</th>
<th>Pravidlo</th>
<th>Typ události</th>
<th>Služba</th>
<th>Incident</th>
<th>Stav</th>
<th>Úloha</th>
<th>Vytvořeno</th>
<th>Akce</th>
</tr>
{render_event_rows(events)}
</table>
</div>
""",
user=user,
)
@router.get("/alerting/events/{event_id}", response_class=HTMLResponse)
def alert_event_detail(event_id: int, request: Request, user=Depends(require_user)):
event = get_alert_event(event_id)
if not event:
raise HTTPException(status_code=404, detail="Alert event nenalezeno")
rule_link = ""
if event.get("rule_id"):
rule_id = html.escape(str(event.get("rule_id")))
rule_name = html.escape(event.get("rule_name") or f"#{rule_id}")
rule_link = f'<a href="/portal/alerting/rules/{rule_id}">{rule_name}</a>'
job_link = ""
if event.get("job_id"):
job_id = html.escape(str(event.get("job_id")))
job_link = f'<a href="/portal/jobs/{job_id}">#{job_id}</a>'
error_text = html.escape(event.get("error_text", "") or "")
payload = html.escape(pretty_json(event.get("payload_json")))
return page(
f"Alert event #{event_id}",
f"""
<div class="card">
<h2>Alert event #{html.escape(str(event_id))}</h2>
<p>
<a class="btn" href="/portal/alerting/events">&larr; Zpět na alert eventy</a>
<a class="btn btn-secondary" href="/portal/alerting/rules">Alert pravidla</a>
</p>
</div>
<div class="card">
<h2>Souhrn</h2>
<table>
<tr><th>Pravidlo</th><td>{rule_link}</td></tr>
<tr><th>Typ události</th><td>{html.escape(event.get("event_type", "") or "")}</td></tr>
<tr><th>Služba</th><td>{html.escape(event.get("service_id", "") or "")}</td></tr>
<tr><th>Incident</th><td>{html.escape(str(event.get("incident_id") or ""))}</td></tr>
<tr><th>Stav</th><td>{render_status(event.get("status"))}</td></tr>
<tr><th>Úloha</th><td>{job_link}</td></tr>
<tr><th>Vytvořeno</th><td>{html.escape(event.get("created_at", "") or "")}</td></tr>
<tr><th>Zpracováno</th><td>{html.escape(event.get("processed_at", "") or "")}</td></tr>
</table>
</div>
<div class="grid">
<div class="card">
<h2>Payload JSON</h2>
<pre class="log-viewer log-stdout">{payload}</pre>
</div>
<div class="card">
<h2>Chyba</h2>
<pre class="log-viewer log-stderr">{error_text}</pre>
</div>
</div>
""",
user=user,
)