diff --git a/app/main.py b/app/main.py index 54b0284..d878f41 100644 --- a/app/main.py +++ b/app/main.py @@ -23,8 +23,6 @@ CORE_SERVICES = { "appfactory-monitor": "http://appfactory-monitor:9300/health", } -TERMINAL_JOB_STATUSES = {"success", "failed", "cancelled"} - app = FastAPI(title="AppFactory Monitor") _monitor_thread = None @@ -40,7 +38,6 @@ def get_connection(): def get_app_services(): con = get_connection() - rows = con.execute( """ SELECT id, health_url @@ -49,7 +46,6 @@ def get_app_services(): ORDER BY id """ ).fetchall() - con.close() services = {} @@ -68,7 +64,6 @@ def get_app_services(): def cleanup_old_health_records(): con = get_connection() - con.execute( """ DELETE FROM service_health @@ -76,7 +71,6 @@ def cleanup_old_health_records(): """, (f"-{HEALTH_RETENTION_DAYS} days",), ) - con.commit() con.close() @@ -111,12 +105,158 @@ def save_health_result( ), ) - health_id = cur.lastrowid - + health_id = int(cur.lastrowid) con.commit() con.close() - return int(health_id) + 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): @@ -135,7 +275,6 @@ def get_open_incident(service_id: str): ).fetchone() con.close() - return row @@ -148,7 +287,7 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str con = get_connection() - con.execute( + cur = con.execute( """ INSERT INTO service_incidents ( service_id, @@ -168,6 +307,14 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str ), ) + incident_id = int(cur.lastrowid) + + metadata = { + "incident_id": incident_id, + "health_id": health_id, + "status": status, + } + con.execute( """ INSERT INTO audit_events ( @@ -191,19 +338,28 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str """, ( service_id, - json.dumps( - { - "health_id": health_id, - "status": status, - }, - ensure_ascii=False, - ), + 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) @@ -211,6 +367,8 @@ def resolve_incident(service_id: str, health_id: int): if not incident: return + incident_id = int(incident["id"]) + con = get_connection() con.execute( @@ -227,10 +385,19 @@ def resolve_incident(service_id: str, health_id: int): """, ( health_id, - incident["id"], + incident_id, ), ) + row = con.execute( + """ + SELECT * + FROM service_incidents + WHERE id = ? + """, + (incident_id,), + ).fetchone() + con.execute( """ INSERT INTO audit_events ( @@ -256,7 +423,7 @@ def resolve_incident(service_id: str, health_id: int): service_id, json.dumps( { - "incident_id": incident["id"], + "incident_id": incident_id, "health_id": health_id, }, ensure_ascii=False, @@ -267,6 +434,22 @@ def resolve_incident(service_id: str, health_id: int): 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":