827 lines
20 KiB
Python
827 lines
20 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from fastapi import FastAPI
|
|
|
|
DB_FILE = Path("/opt/appfactory/data/appfactory/appfactory.db")
|
|
|
|
CHECK_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_MONITOR_INTERVAL_SECONDS", "30"))
|
|
REQUEST_TIMEOUT_SECONDS = int(os.getenv("APPFACTORY_MONITOR_TIMEOUT_SECONDS", "5"))
|
|
HEALTH_RETENTION_DAYS = int(os.getenv("APPFACTORY_HEALTH_RETENTION_DAYS", "7"))
|
|
SCHEDULER_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_SCHEDULER_INTERVAL_SECONDS", "60"))
|
|
|
|
CORE_SERVICES = {
|
|
"appfactory-portal": "http://appfactory-portal:9100/health",
|
|
"appfactory-webhook": "http://appfactory-webhook:9000/health",
|
|
"appfactory-worker": "http://appfactory-worker:9200/health",
|
|
"appfactory-monitor": "http://appfactory-monitor:9300/health",
|
|
}
|
|
|
|
app = FastAPI(title="AppFactory Monitor")
|
|
|
|
_monitor_thread = None
|
|
_scheduler_thread = None
|
|
_monitor_started = False
|
|
|
|
|
|
def get_connection():
|
|
con = sqlite3.connect(DB_FILE, timeout=30)
|
|
con.row_factory = sqlite3.Row
|
|
return con
|
|
|
|
|
|
def get_app_services():
|
|
con = get_connection()
|
|
rows = con.execute(
|
|
"""
|
|
SELECT id, health_url
|
|
FROM apps
|
|
WHERE is_enabled = 1
|
|
ORDER BY id
|
|
"""
|
|
).fetchall()
|
|
con.close()
|
|
|
|
services = {}
|
|
|
|
for row in rows:
|
|
app_id = row["id"]
|
|
health_url = row["health_url"] or "/health"
|
|
|
|
if health_url.startswith("https://appfactory-caddy"):
|
|
health_url = health_url.replace(
|
|
"https://appfactory-caddy",
|
|
"http://appfactory-caddy",
|
|
1,
|
|
)
|
|
|
|
if health_url.startswith("http://") or health_url.startswith("https://"):
|
|
services[app_id] = health_url
|
|
else:
|
|
services[app_id] = f"http://appfactory-caddy/apps/{app_id}{health_url}"
|
|
|
|
return services
|
|
|
|
|
|
def cleanup_old_health_records():
|
|
con = get_connection()
|
|
con.execute(
|
|
"""
|
|
DELETE FROM service_health
|
|
WHERE checked_at < datetime('now', ?)
|
|
""",
|
|
(f"-{HEALTH_RETENTION_DAYS} days",),
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def save_health_result(
|
|
service_id: str,
|
|
status: str,
|
|
http_status: int | None,
|
|
response_time_ms: int | None,
|
|
error_text: str | None,
|
|
) -> int:
|
|
con = get_connection()
|
|
|
|
cur = con.execute(
|
|
"""
|
|
INSERT INTO service_health (
|
|
service_id,
|
|
status,
|
|
http_status,
|
|
response_time_ms,
|
|
error_text,
|
|
checked_at
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
|
""",
|
|
(
|
|
service_id,
|
|
status,
|
|
http_status,
|
|
response_time_ms,
|
|
error_text,
|
|
),
|
|
)
|
|
|
|
health_id = int(cur.lastrowid)
|
|
con.commit()
|
|
con.close()
|
|
|
|
return health_id
|
|
|
|
|
|
def enqueue_alerts(
|
|
event_type: str,
|
|
service_id: str,
|
|
incident_id: int | None,
|
|
payload: dict,
|
|
):
|
|
con = get_connection()
|
|
|
|
rules = con.execute(
|
|
"""
|
|
SELECT *
|
|
FROM alert_rules
|
|
WHERE is_enabled = 1
|
|
AND event_type = ?
|
|
AND (service_id IS NULL OR service_id = '' OR service_id = ?)
|
|
ORDER BY id
|
|
""",
|
|
(
|
|
event_type,
|
|
service_id,
|
|
),
|
|
).fetchall()
|
|
|
|
for rule in rules:
|
|
alert_payload = dict(payload)
|
|
alert_payload["rule_id"] = rule["id"]
|
|
alert_payload["rule_name"] = rule["name"]
|
|
alert_payload["script_name"] = rule["script_name"]
|
|
|
|
alert_cur = con.execute(
|
|
"""
|
|
INSERT INTO alert_events (
|
|
rule_id,
|
|
event_type,
|
|
service_id,
|
|
incident_id,
|
|
status,
|
|
payload_json,
|
|
created_at
|
|
)
|
|
VALUES (?, ?, ?, ?, 'queued', ?, CURRENT_TIMESTAMP)
|
|
""",
|
|
(
|
|
rule["id"],
|
|
event_type,
|
|
service_id,
|
|
incident_id,
|
|
json.dumps(alert_payload, ensure_ascii=False),
|
|
),
|
|
)
|
|
|
|
alert_event_id = int(alert_cur.lastrowid)
|
|
|
|
job_payload = {
|
|
"alert_event_id": alert_event_id,
|
|
"rule_id": rule["id"],
|
|
"script_name": rule["script_name"],
|
|
"event_type": event_type,
|
|
"service_id": service_id,
|
|
"incident_id": incident_id,
|
|
"alert_payload": alert_payload,
|
|
}
|
|
|
|
job_cur = con.execute(
|
|
"""
|
|
INSERT INTO jobs (
|
|
type,
|
|
target_type,
|
|
target_id,
|
|
status,
|
|
source,
|
|
created_by_username,
|
|
payload_json,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
'run_alert_script',
|
|
'alert_rule',
|
|
?,
|
|
'queued',
|
|
'alerting',
|
|
'system',
|
|
?,
|
|
CURRENT_TIMESTAMP
|
|
)
|
|
""",
|
|
(
|
|
rule["script_name"],
|
|
json.dumps(job_payload, ensure_ascii=False),
|
|
),
|
|
)
|
|
|
|
job_id = int(job_cur.lastrowid)
|
|
|
|
con.execute(
|
|
"""
|
|
UPDATE alert_events
|
|
SET job_id = ?
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
job_id,
|
|
alert_event_id,
|
|
),
|
|
)
|
|
|
|
con.execute(
|
|
"""
|
|
INSERT INTO audit_events (
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
'system',
|
|
'alert.enqueued',
|
|
'alert_rule',
|
|
?,
|
|
'monitor',
|
|
?,
|
|
CURRENT_TIMESTAMP
|
|
)
|
|
""",
|
|
(
|
|
str(rule["id"]),
|
|
json.dumps(
|
|
{
|
|
"alert_event_id": alert_event_id,
|
|
"job_id": job_id,
|
|
"event_type": event_type,
|
|
"service_id": service_id,
|
|
"incident_id": incident_id,
|
|
"script_name": rule["script_name"],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
),
|
|
)
|
|
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def get_open_incident(service_id: str):
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT *
|
|
FROM service_incidents
|
|
WHERE service_id = ?
|
|
AND status = 'open'
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
""",
|
|
(service_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return row
|
|
|
|
|
|
def open_incident(service_id: str, health_id: int, status: str, error_text: str | None):
|
|
if get_open_incident(service_id):
|
|
return
|
|
|
|
title = f"Služba {service_id} není zdravá"
|
|
description = error_text or f"Health status: {status}"
|
|
|
|
con = get_connection()
|
|
|
|
cur = con.execute(
|
|
"""
|
|
INSERT INTO service_incidents (
|
|
service_id,
|
|
status,
|
|
started_at,
|
|
start_health_id,
|
|
title,
|
|
description
|
|
)
|
|
VALUES (?, 'open', CURRENT_TIMESTAMP, ?, ?, ?)
|
|
""",
|
|
(
|
|
service_id,
|
|
health_id,
|
|
title,
|
|
description,
|
|
),
|
|
)
|
|
|
|
incident_id = int(cur.lastrowid)
|
|
|
|
metadata = {
|
|
"incident_id": incident_id,
|
|
"health_id": health_id,
|
|
"status": status,
|
|
}
|
|
|
|
con.execute(
|
|
"""
|
|
INSERT INTO audit_events (
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
'system',
|
|
'service.incident.opened',
|
|
'service',
|
|
?,
|
|
'monitor',
|
|
?,
|
|
CURRENT_TIMESTAMP
|
|
)
|
|
""",
|
|
(
|
|
service_id,
|
|
json.dumps(metadata, ensure_ascii=False),
|
|
),
|
|
)
|
|
|
|
con.commit()
|
|
con.close()
|
|
|
|
enqueue_alerts(
|
|
event_type="incident.opened",
|
|
service_id=service_id,
|
|
incident_id=incident_id,
|
|
payload={
|
|
"incident_id": incident_id,
|
|
"service_id": service_id,
|
|
"health_id": health_id,
|
|
"health_status": status,
|
|
"title": title,
|
|
"description": description,
|
|
"error_text": error_text,
|
|
},
|
|
)
|
|
|
|
|
|
def resolve_incident(service_id: str, health_id: int):
|
|
incident = get_open_incident(service_id)
|
|
|
|
if not incident:
|
|
return
|
|
|
|
incident_id = int(incident["id"])
|
|
|
|
con = get_connection()
|
|
|
|
con.execute(
|
|
"""
|
|
UPDATE service_incidents
|
|
SET status = 'resolved',
|
|
ended_at = CURRENT_TIMESTAMP,
|
|
end_health_id = ?,
|
|
duration_seconds = CAST(
|
|
(julianday(CURRENT_TIMESTAMP) - julianday(started_at)) * 86400
|
|
AS INTEGER
|
|
)
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
health_id,
|
|
incident_id,
|
|
),
|
|
)
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT *
|
|
FROM service_incidents
|
|
WHERE id = ?
|
|
""",
|
|
(incident_id,),
|
|
).fetchone()
|
|
|
|
con.execute(
|
|
"""
|
|
INSERT INTO audit_events (
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
'system',
|
|
'service.incident.resolved',
|
|
'service',
|
|
?,
|
|
'monitor',
|
|
?,
|
|
CURRENT_TIMESTAMP
|
|
)
|
|
""",
|
|
(
|
|
service_id,
|
|
json.dumps(
|
|
{
|
|
"incident_id": incident_id,
|
|
"health_id": health_id,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
),
|
|
)
|
|
|
|
con.commit()
|
|
con.close()
|
|
|
|
enqueue_alerts(
|
|
event_type="incident.resolved",
|
|
service_id=service_id,
|
|
incident_id=incident_id,
|
|
payload={
|
|
"incident_id": incident_id,
|
|
"service_id": service_id,
|
|
"health_id": health_id,
|
|
"title": row["title"] if row else None,
|
|
"description": row["description"] if row else None,
|
|
"started_at": row["started_at"] if row else None,
|
|
"ended_at": row["ended_at"] if row else None,
|
|
"duration_seconds": row["duration_seconds"] if row else None,
|
|
},
|
|
)
|
|
|
|
|
|
def update_incident_state(service_id: str, health_id: int, status: str, error_text: str | None):
|
|
if status == "healthy":
|
|
resolve_incident(service_id, health_id)
|
|
else:
|
|
open_incident(service_id, health_id, status, error_text)
|
|
|
|
|
|
def check_service(service_id: str, url: str):
|
|
start = time.monotonic()
|
|
|
|
try:
|
|
response = requests.get(
|
|
url,
|
|
timeout=REQUEST_TIMEOUT_SECONDS,
|
|
)
|
|
|
|
elapsed_ms = int((time.monotonic() - start) * 1000)
|
|
|
|
if 200 <= response.status_code < 300:
|
|
health_id = save_health_result(
|
|
service_id=service_id,
|
|
status="healthy",
|
|
http_status=response.status_code,
|
|
response_time_ms=elapsed_ms,
|
|
error_text=None,
|
|
)
|
|
|
|
update_incident_state(
|
|
service_id=service_id,
|
|
health_id=health_id,
|
|
status="healthy",
|
|
error_text=None,
|
|
)
|
|
else:
|
|
error_text = response.text[:500]
|
|
|
|
health_id = save_health_result(
|
|
service_id=service_id,
|
|
status="unhealthy",
|
|
http_status=response.status_code,
|
|
response_time_ms=elapsed_ms,
|
|
error_text=error_text,
|
|
)
|
|
|
|
update_incident_state(
|
|
service_id=service_id,
|
|
health_id=health_id,
|
|
status="unhealthy",
|
|
error_text=error_text,
|
|
)
|
|
|
|
except Exception as exc:
|
|
elapsed_ms = int((time.monotonic() - start) * 1000)
|
|
error_text = str(exc)
|
|
|
|
health_id = save_health_result(
|
|
service_id=service_id,
|
|
status="unreachable",
|
|
http_status=None,
|
|
response_time_ms=elapsed_ms,
|
|
error_text=error_text,
|
|
)
|
|
|
|
update_incident_state(
|
|
service_id=service_id,
|
|
health_id=health_id,
|
|
status="unreachable",
|
|
error_text=error_text,
|
|
)
|
|
|
|
|
|
def run_check_cycle():
|
|
cleanup_old_health_records()
|
|
|
|
services = {}
|
|
services.update(CORE_SERVICES)
|
|
services.update(get_app_services())
|
|
|
|
for service_id, url in services.items():
|
|
check_service(service_id, url)
|
|
|
|
|
|
def parse_time_hhmm(value: str | None) -> tuple[int, int]:
|
|
if not value:
|
|
return 3, 0
|
|
|
|
parts = value.strip().split(":", 1)
|
|
|
|
if len(parts) != 2:
|
|
return 3, 0
|
|
|
|
try:
|
|
hour = int(parts[0])
|
|
minute = int(parts[1])
|
|
except ValueError:
|
|
return 3, 0
|
|
|
|
if hour < 0 or hour > 23:
|
|
hour = 3
|
|
|
|
if minute < 0 or minute > 59:
|
|
minute = 0
|
|
|
|
return hour, minute
|
|
|
|
|
|
def compute_next_run_at(schedule_type: str, schedule_time: str | None) -> str:
|
|
now = datetime.utcnow().replace(microsecond=0)
|
|
schedule_type = (schedule_type or "daily").strip().lower()
|
|
hour, minute = parse_time_hhmm(schedule_time)
|
|
|
|
if schedule_type == "hourly":
|
|
candidate = now.replace(minute=minute, second=0)
|
|
if candidate <= now:
|
|
candidate += timedelta(hours=1)
|
|
return candidate.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
if schedule_type == "weekly":
|
|
candidate = now.replace(hour=hour, minute=minute, second=0)
|
|
while candidate <= now:
|
|
candidate += timedelta(days=7)
|
|
return candidate.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
if schedule_type == "monthly":
|
|
candidate = now.replace(day=1, hour=hour, minute=minute, second=0)
|
|
if candidate <= now:
|
|
if candidate.month == 12:
|
|
candidate = candidate.replace(year=candidate.year + 1, month=1)
|
|
else:
|
|
candidate = candidate.replace(month=candidate.month + 1)
|
|
return candidate.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
candidate = now.replace(hour=hour, minute=minute, second=0)
|
|
if candidate <= now:
|
|
candidate += timedelta(days=1)
|
|
|
|
return candidate.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def sync_scheduled_script_results():
|
|
con = get_connection()
|
|
|
|
rows = con.execute(
|
|
"""
|
|
SELECT
|
|
scheduled_scripts.id AS scheduled_script_id,
|
|
scheduled_scripts.last_job_id AS last_job_id,
|
|
jobs.status AS job_status,
|
|
jobs.error_text AS job_error_text,
|
|
jobs.started_at AS job_started_at,
|
|
jobs.finished_at AS job_finished_at
|
|
FROM scheduled_scripts
|
|
JOIN jobs ON jobs.id = scheduled_scripts.last_job_id
|
|
WHERE scheduled_scripts.is_running = 1
|
|
AND jobs.status IN ('success', 'failed', 'cancelled')
|
|
"""
|
|
).fetchall()
|
|
|
|
for row in rows:
|
|
last_run_at = row["job_finished_at"] or row["job_started_at"]
|
|
|
|
con.execute(
|
|
"""
|
|
UPDATE scheduled_scripts
|
|
SET is_running = 0,
|
|
last_status = ?,
|
|
last_error_text = ?,
|
|
last_run_at = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
row["job_status"],
|
|
row["job_error_text"],
|
|
last_run_at,
|
|
row["scheduled_script_id"],
|
|
),
|
|
)
|
|
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def enqueue_due_scheduled_scripts():
|
|
con = get_connection()
|
|
|
|
con.execute("BEGIN IMMEDIATE")
|
|
|
|
rows = con.execute(
|
|
"""
|
|
SELECT *
|
|
FROM scheduled_scripts
|
|
WHERE is_enabled = 1
|
|
AND is_running = 0
|
|
AND next_run_at IS NOT NULL
|
|
AND next_run_at <= CURRENT_TIMESTAMP
|
|
ORDER BY next_run_at ASC, id ASC
|
|
"""
|
|
).fetchall()
|
|
|
|
for row in rows:
|
|
script_id = int(row["id"])
|
|
script_name = row["script_name"]
|
|
timeout_seconds = int(row["timeout_seconds"] or 300)
|
|
next_run_at = compute_next_run_at(row["schedule_type"], row["schedule_time"])
|
|
|
|
payload = {
|
|
"scheduled_script_id": script_id,
|
|
"script_name": script_name,
|
|
"timeout_seconds": timeout_seconds,
|
|
}
|
|
|
|
cur = con.execute(
|
|
"""
|
|
INSERT INTO jobs (
|
|
type,
|
|
target_type,
|
|
target_id,
|
|
status,
|
|
source,
|
|
created_by_username,
|
|
payload_json,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
'run_script',
|
|
'scheduled_script',
|
|
?,
|
|
'queued',
|
|
'scheduler',
|
|
'system',
|
|
?,
|
|
CURRENT_TIMESTAMP
|
|
)
|
|
""",
|
|
(
|
|
script_name,
|
|
json.dumps(payload, ensure_ascii=False),
|
|
),
|
|
)
|
|
|
|
job_id = int(cur.lastrowid)
|
|
|
|
con.execute(
|
|
"""
|
|
UPDATE scheduled_scripts
|
|
SET is_running = 1,
|
|
last_job_id = ?,
|
|
last_status = 'queued',
|
|
last_error_text = NULL,
|
|
next_run_at = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
job_id,
|
|
next_run_at,
|
|
script_id,
|
|
),
|
|
)
|
|
|
|
con.execute(
|
|
"""
|
|
INSERT INTO audit_events (
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
'system',
|
|
'scheduled_script.enqueued',
|
|
'scheduled_script',
|
|
?,
|
|
'scheduler',
|
|
?,
|
|
CURRENT_TIMESTAMP
|
|
)
|
|
""",
|
|
(
|
|
str(script_id),
|
|
json.dumps(
|
|
{
|
|
"job_id": job_id,
|
|
"script_name": script_name,
|
|
"next_run_at": next_run_at,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
),
|
|
)
|
|
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def run_scheduler_cycle():
|
|
sync_scheduled_script_results()
|
|
enqueue_due_scheduled_scripts()
|
|
|
|
|
|
def monitor_loop():
|
|
while True:
|
|
try:
|
|
run_check_cycle()
|
|
except Exception as exc:
|
|
print(f"Monitor loop error: {exc}", flush=True)
|
|
|
|
time.sleep(CHECK_INTERVAL_SECONDS)
|
|
|
|
|
|
def scheduler_loop():
|
|
while True:
|
|
try:
|
|
run_scheduler_cycle()
|
|
except Exception as exc:
|
|
print(f"Scheduler loop error: {exc}", flush=True)
|
|
|
|
time.sleep(SCHEDULER_INTERVAL_SECONDS)
|
|
|
|
|
|
@app.on_event("startup")
|
|
def start_monitor():
|
|
global _monitor_thread
|
|
global _scheduler_thread
|
|
global _monitor_started
|
|
|
|
if _monitor_started:
|
|
return
|
|
|
|
_monitor_started = True
|
|
|
|
_monitor_thread = threading.Thread(
|
|
target=monitor_loop,
|
|
daemon=True,
|
|
)
|
|
_monitor_thread.start()
|
|
|
|
_scheduler_thread = threading.Thread(
|
|
target=scheduler_loop,
|
|
daemon=True,
|
|
)
|
|
_scheduler_thread.start()
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {
|
|
"status": "ok",
|
|
"interval_seconds": CHECK_INTERVAL_SECONDS,
|
|
"timeout_seconds": REQUEST_TIMEOUT_SECONDS,
|
|
"retention_days": HEALTH_RETENTION_DAYS,
|
|
"scheduler_interval_seconds": SCHEDULER_INTERVAL_SECONDS,
|
|
}
|
|
|
|
|
|
@app.get("/services")
|
|
def services():
|
|
result = {}
|
|
result.update(CORE_SERVICES)
|
|
result.update(get_app_services())
|
|
|
|
return result
|