507 lines
21 KiB
Python
507 lines
21 KiB
Python
"""
|
||
MERICA Lakehouse Alerting & Notification Hook for Apache Airflow 3.
|
||
Provides multi-channel incident dispatching (Telegram, Slack, Discord, SITOMAT, Email),
|
||
governance metadata enrichment (Domain, Owner, Tier), and anti-alert fatigue throttling.
|
||
Adheres to Dokumen 34 Architectural Blueprint.
|
||
"""
|
||
import os
|
||
import json
|
||
import time
|
||
import smtplib
|
||
import logging
|
||
import urllib.request
|
||
import urllib.error
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple, Union
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Global In-Memory Cache for Anti-Alert Fatigue Coalescing
|
||
_ALERT_SUPPRESSION_CACHE: Dict[str, Dict[str, Any]] = {}
|
||
|
||
|
||
class MericaAlertHook:
|
||
"""
|
||
Enterprise Alert Dispatcher with Governance Awareness and Anti-Alert Fatigue Engine.
|
||
Dispatches rich card notifications across multiple channels on task failure,
|
||
data quality breaches, and SLA misses.
|
||
"""
|
||
|
||
DEFAULT_COALESCING_WINDOW = 300 # 5 minutes in seconds
|
||
|
||
@staticmethod
|
||
def send_webhook(
|
||
webhook_url: str,
|
||
payload: Dict[str, Any],
|
||
timeout: int = 8,
|
||
headers: Optional[Dict[str, str]] = None
|
||
) -> bool:
|
||
"""
|
||
Kirim payload JSON ke target Webhook dengan timeout dan error handling aman.
|
||
"""
|
||
if not webhook_url:
|
||
logger.warning("Empty webhook URL provided to MericaAlertHook")
|
||
return False
|
||
|
||
req_headers = {"Content-Type": "application/json"}
|
||
if headers:
|
||
req_headers.update(headers)
|
||
|
||
try:
|
||
req = urllib.request.Request(
|
||
webhook_url,
|
||
data=json.dumps(payload).encode("utf-8"),
|
||
headers=req_headers,
|
||
method="POST"
|
||
)
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
status_code = getattr(resp, "status", 200)
|
||
if status_code in (200, 201, 204):
|
||
logger.info("Successfully dispatched alert webhook to %s (Status: %s)", webhook_url, status_code)
|
||
return True
|
||
logger.warning("Alert webhook returned unexpected status %s: %s", status_code, webhook_url)
|
||
return False
|
||
except Exception as err:
|
||
logger.error("Failed to dispatch alert webhook to %s: %s", webhook_url, err)
|
||
return False
|
||
|
||
@classmethod
|
||
def check_and_record_suppression(
|
||
cls,
|
||
cache_key: str,
|
||
window_seconds: int = DEFAULT_COALESCING_WINDOW
|
||
) -> Tuple[bool, int]:
|
||
"""
|
||
Anti-Alert Fatigue Engine:
|
||
Memeriksa apakah insiden serupa telah dikirim dalam jendela waktu tertentu.
|
||
Mengembalikan (is_suppressed, suppression_count).
|
||
"""
|
||
now = time.time()
|
||
record = _ALERT_SUPPRESSION_CACHE.get(cache_key)
|
||
|
||
if record is None:
|
||
_ALERT_SUPPRESSION_CACHE[cache_key] = {"first_seen": now, "last_seen": now, "count": 1}
|
||
return False, 1
|
||
|
||
elapsed = now - record["first_seen"]
|
||
if elapsed < window_seconds:
|
||
record["count"] += 1
|
||
record["last_seen"] = now
|
||
return True, record["count"]
|
||
|
||
# Jendela waktu terlampaui, reset cache
|
||
_ALERT_SUPPRESSION_CACHE[cache_key] = {"first_seen": now, "last_seen": now, "count": 1}
|
||
return False, 1
|
||
|
||
@classmethod
|
||
def clear_suppression_cache(cls):
|
||
"""Utility for test suites to clear the suppression cache."""
|
||
_ALERT_SUPPRESSION_CACHE.clear()
|
||
|
||
@classmethod
|
||
def send_telegram_alert(
|
||
cls,
|
||
bot_token: str,
|
||
chat_id: str,
|
||
message: str,
|
||
parse_mode: str = "HTML"
|
||
) -> bool:
|
||
"""Kirim pesan notifikasi ke Telegram Bot."""
|
||
if not bot_token or not chat_id:
|
||
logger.debug("Telegram credentials not configured; skipping Telegram alert.")
|
||
return False
|
||
|
||
tg_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
|
||
payload = {
|
||
"chat_id": chat_id,
|
||
"text": message,
|
||
"parse_mode": parse_mode,
|
||
"disable_web_page_preview": True
|
||
}
|
||
return cls.send_webhook(tg_url, payload)
|
||
|
||
@classmethod
|
||
def send_slack_alert(cls, webhook_url: str, payload: Dict[str, Any]) -> bool:
|
||
"""Kirim pesan Rich Block Kit ke Slack / Mattermost Webhook."""
|
||
if not webhook_url:
|
||
return False
|
||
return cls.send_webhook(webhook_url, payload)
|
||
|
||
@classmethod
|
||
def send_discord_alert(cls, webhook_url: str, payload: Dict[str, Any]) -> bool:
|
||
"""Kirim pesan Rich Embed ke Discord Webhook."""
|
||
if not webhook_url:
|
||
return False
|
||
return cls.send_webhook(webhook_url, payload)
|
||
|
||
@classmethod
|
||
def send_sitomat_alert(
|
||
cls,
|
||
endpoint_url: str,
|
||
incident_payload: Dict[str, Any],
|
||
api_key: Optional[str] = None
|
||
) -> bool:
|
||
"""Kirim laporan insiden ke SITOMAT F16 API Gateway."""
|
||
if not endpoint_url:
|
||
return False
|
||
headers = {}
|
||
if api_key:
|
||
headers["X-API-Key"] = api_key
|
||
headers["Authorization"] = f"Bearer {api_key}"
|
||
return cls.send_webhook(endpoint_url, incident_payload, headers=headers)
|
||
|
||
@classmethod
|
||
def send_email_alert(
|
||
cls,
|
||
smtp_host: str,
|
||
smtp_port: int,
|
||
sender: str,
|
||
recipients: List[str],
|
||
subject: str,
|
||
body_html: str,
|
||
username: Optional[str] = None,
|
||
password: Optional[str] = None,
|
||
use_tls: bool = True
|
||
) -> bool:
|
||
"""Kirim email notifikasi insiden via SMTP Enterprise."""
|
||
if not smtp_host or not sender or not recipients:
|
||
logger.debug("SMTP configuration incomplete; skipping email alert.")
|
||
return False
|
||
|
||
try:
|
||
msg = MIMEMultipart("alternative")
|
||
msg["Subject"] = subject
|
||
msg["From"] = sender
|
||
msg["To"] = ", ".join(recipients)
|
||
msg.attach(MIMEText(body_html, "html"))
|
||
|
||
with smtplib.SMTP(smtp_host, smtp_port, timeout=10) as server:
|
||
if use_tls:
|
||
server.starttls()
|
||
if username and password:
|
||
server.login(username, password)
|
||
server.sendmail(sender, recipients, msg.as_string())
|
||
logger.info("Successfully sent incident email to %s", recipients)
|
||
return True
|
||
except Exception as err:
|
||
logger.error("Failed to send SMTP email alert: %s", err)
|
||
return False
|
||
|
||
@classmethod
|
||
def extract_context(cls, context: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
Mengekstrak informasi runtime dan memperkaya dengan metadata tata kelola
|
||
(Domain, Owner, Tier, SLA) dari context Airflow Task / DAG.
|
||
"""
|
||
ti = context.get("task_instance")
|
||
dag = context.get("dag")
|
||
exception = context.get("exception")
|
||
|
||
dag_id = dag.dag_id if dag else (ti.dag_id if ti else "unknown_pipeline")
|
||
task_id = ti.task_id if ti else "unknown_task"
|
||
try_number = ti.try_number if ti else 1
|
||
max_tries = getattr(ti, "max_tries", 1) if ti else 1
|
||
duration_sec = getattr(ti, "duration", None)
|
||
if isinstance(duration_sec, (int, float)):
|
||
duration_str = f"{duration_sec:.2f}s"
|
||
elif duration_sec is not None and not str(duration_sec).startswith("<MagicMock"):
|
||
duration_str = f"{duration_sec}s"
|
||
else:
|
||
duration_str = "--"
|
||
|
||
# Log URL
|
||
log_url = getattr(ti, "log_url", "") if ti else ""
|
||
if not log_url:
|
||
airflow_url = os.getenv("AIRFLOW_PUBLIC_URL", "http://localhost:8088")
|
||
log_url = f"{airflow_url}/dags/{dag_id}/grid"
|
||
|
||
# Governance Metadata dari DAG params / default
|
||
dag_params = getattr(dag, "params", {})
|
||
if isinstance(dag_params, dict):
|
||
domain = dag_params.get("domain", os.getenv("GOVERNANCE_DEFAULT_DOMAIN", "Lingkungan Hidup & Kehutanan"))
|
||
owner = dag_params.get("owner", os.getenv("GOVERNANCE_DEFAULT_OWNER", "dlh_data_engineering_group"))
|
||
tier = dag_params.get("tier", os.getenv("GOVERNANCE_DEFAULT_TIER", "Tier.Tier2"))
|
||
else:
|
||
domain = os.getenv("GOVERNANCE_DEFAULT_DOMAIN", "Lingkungan Hidup & Kehutanan")
|
||
owner = os.getenv("GOVERNANCE_DEFAULT_OWNER", "dlh_data_engineering_group")
|
||
tier = os.getenv("GOVERNANCE_DEFAULT_TIER", "Tier.Tier2")
|
||
|
||
# Format error trace
|
||
if exception:
|
||
err_msg = str(exception)
|
||
else:
|
||
err_msg = "Task exited with error or failed quality assertion."
|
||
# Truncate error message if too long
|
||
if len(err_msg) > 350:
|
||
err_msg = err_msg[:350] + "... (truncated)"
|
||
|
||
# Tentukan severity P0/P1/P2 selaras tier
|
||
if "tier1" in str(tier).lower():
|
||
severity = "🔴 P0 (CRITICAL)"
|
||
color_hex = "#E01E5A"
|
||
elif "tier2" in str(tier).lower():
|
||
severity = "⚠️ P1 (WARNING)"
|
||
color_hex = "#ECB22E"
|
||
else:
|
||
severity = "ℹ️ P2 (INFO)"
|
||
color_hex = "#2EB886"
|
||
|
||
return {
|
||
"dag_id": dag_id,
|
||
"task_id": task_id,
|
||
"try_number": try_number,
|
||
"max_tries": max_tries,
|
||
"duration": duration_str,
|
||
"log_url": log_url,
|
||
"error_msg": err_msg,
|
||
"domain": domain,
|
||
"owner": owner,
|
||
"tier": tier,
|
||
"severity": severity,
|
||
"color_hex": color_hex,
|
||
"timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
|
||
}
|
||
|
||
@classmethod
|
||
def dispatch_alert(
|
||
cls,
|
||
incident_type: str,
|
||
ctx_data: Dict[str, Any],
|
||
alert_config: Optional[Dict[str, Any]] = None
|
||
) -> Dict[str, bool]:
|
||
"""
|
||
Pusat perutean notifikasi multi-saluran dengan verifikasi Anti-Alert Fatigue.
|
||
"""
|
||
dag_id = ctx_data["dag_id"]
|
||
task_id = ctx_data["task_id"]
|
||
cache_key = f"{dag_id}:{task_id}:{incident_type}"
|
||
|
||
coalescing_window = (
|
||
alert_config.get("coalescing_window_seconds", cls.DEFAULT_COALESCING_WINDOW)
|
||
if alert_config else cls.DEFAULT_COALESCING_WINDOW
|
||
)
|
||
|
||
is_suppressed, count = cls.check_and_record_suppression(cache_key, coalescing_window)
|
||
if is_suppressed:
|
||
logger.warning(
|
||
"Alert throttled by Anti-Alert Fatigue Engine for key '%s' (Suppressed count: %s)",
|
||
cache_key, count
|
||
)
|
||
return {"throttled": True, "suppression_count": count}
|
||
|
||
results: Dict[str, bool] = {}
|
||
|
||
om_public_url = os.getenv("OPENMETADATA_PUBLIC_URL", os.getenv("OPENMETADATA_HOST", "http://localhost:14585"))
|
||
pipeline_svc = os.getenv("OPENMETADATA_PIPELINE_SERVICE_NAME", "dlh_airflow_service")
|
||
|
||
# 1. Telegram Dispatch
|
||
tg_bot_token = os.getenv("ALERT_TELEGRAM_BOT_TOKEN")
|
||
tg_chat_id = os.getenv("ALERT_TELEGRAM_CHAT_ID")
|
||
if alert_config and "channels" in alert_config:
|
||
tg_conf = alert_config["channels"].get("telegram", {})
|
||
if tg_conf.get("enabled", False):
|
||
tg_bot_token = os.getenv(tg_conf.get("bot_token_secret_key", "ALERT_TELEGRAM_BOT_TOKEN"), tg_bot_token)
|
||
tg_chat_id = os.getenv(tg_conf.get("chat_id_secret_key", "ALERT_TELEGRAM_CHAT_ID"), tg_chat_id)
|
||
|
||
if tg_bot_token and tg_chat_id:
|
||
tg_message = (
|
||
f"<b>{ctx_data['severity']} MERICA PIPELINE ALERT</b>\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"📋 <b>Pipeline:</b> <code>{ctx_data['dag_id']}</code>\n"
|
||
f"⚙️ <b>Task:</b> <code>{ctx_data['task_id']}</code> (Try {ctx_data['try_number']}/{ctx_data['max_tries']})\n"
|
||
f"⏱️ <b>Duration:</b> {ctx_data['duration']}\n"
|
||
f"🏛️ <b>Domain:</b> {ctx_data['domain']}\n"
|
||
f"👤 <b>Owner:</b> <code>{ctx_data['owner']}</code> | <b>Tier:</b> <code>{ctx_data['tier']}</code>\n\n"
|
||
f"⚠️ <b>Root Cause:</b>\n<code>{ctx_data['error_msg']}</code>\n\n"
|
||
f"🔗 <a href='{ctx_data['log_url']}'>Airflow Log</a> | "
|
||
f"<a href='{om_public_url}/pipeline/{pipeline_svc}.{ctx_data['dag_id']}/lineage'>OpenMetadata Lineage</a>"
|
||
)
|
||
results["telegram"] = cls.send_telegram_alert(tg_bot_token, tg_chat_id, tg_message)
|
||
|
||
# 2. Slack Dispatch
|
||
slack_url = os.getenv("ALERT_SLACK_WEBHOOK_URL")
|
||
if alert_config and "channels" in alert_config:
|
||
slack_conf = alert_config["channels"].get("slack", {})
|
||
if slack_conf.get("enabled", False):
|
||
slack_url = os.getenv(slack_conf.get("webhook_url_secret_key", "ALERT_SLACK_WEBHOOK_URL"), slack_url)
|
||
|
||
if slack_url:
|
||
slack_payload = {
|
||
"text": f"{ctx_data['severity']} Pipeline Incident: {ctx_data['dag_id']} / {ctx_data['task_id']}",
|
||
"attachments": [
|
||
{
|
||
"color": ctx_data["color_hex"],
|
||
"blocks": [
|
||
{
|
||
"type": "header",
|
||
"text": {"type": "plain_text", "text": f"{ctx_data['severity']} - {incident_type}"}
|
||
},
|
||
{
|
||
"type": "section",
|
||
"fields": [
|
||
{"type": "mrkdwn", "text": f"*Pipeline:*\n`{ctx_data['dag_id']}`"},
|
||
{"type": "mrkdwn", "text": f"*Task:*\n`{ctx_data['task_id']}`"},
|
||
{"type": "mrkdwn", "text": f"*Domain:*\n{ctx_data['domain']}"},
|
||
{"type": "mrkdwn", "text": f"*Owner:*\n`{ctx_data['owner']}`"},
|
||
{"type": "mrkdwn", "text": f"*Tier:*\n`{ctx_data['tier']}`"},
|
||
{"type": "mrkdwn", "text": f"*Duration:*\n{ctx_data['duration']}"}
|
||
]
|
||
},
|
||
{
|
||
"type": "section",
|
||
"text": {"type": "mrkdwn", "text": f"*Error Message:*\n```{ctx_data['error_msg']}```"}
|
||
},
|
||
{
|
||
"type": "actions",
|
||
"elements": [
|
||
{"type": "button", "text": {"type": "plain_text", "text": "View Airflow Log"}, "url": ctx_data["log_url"]},
|
||
{"type": "button", "text": {"type": "plain_text", "text": "OpenMetadata Lineage"}, "url": f"{om_public_url}/pipeline/{pipeline_svc}.{ctx_data['dag_id']}/lineage"}
|
||
]
|
||
}
|
||
]
|
||
}
|
||
]
|
||
}
|
||
results["slack"] = cls.send_slack_alert(slack_url, slack_payload)
|
||
|
||
# 3. Discord Dispatch
|
||
discord_url = os.getenv("ALERT_DISCORD_WEBHOOK_URL")
|
||
if alert_config and "channels" in alert_config:
|
||
discord_conf = alert_config["channels"].get("discord", {})
|
||
if discord_conf.get("enabled", False):
|
||
discord_url = os.getenv(discord_conf.get("webhook_url_secret_key", "ALERT_DISCORD_WEBHOOK_URL"), discord_url)
|
||
|
||
if discord_url:
|
||
discord_payload = {
|
||
"content": f"{ctx_data['severity']} Pipeline Incident Detected",
|
||
"embeds": [
|
||
{
|
||
"title": f"MERICA Incident: {ctx_data['dag_id']} / {ctx_data['task_id']}",
|
||
"description": f"```{ctx_data['error_msg']}```",
|
||
"color": int(ctx_data["color_hex"].replace("#", ""), 16),
|
||
"fields": [
|
||
{"name": "Domain", "value": ctx_data["domain"], "inline": True},
|
||
{"name": "Owner", "value": ctx_data["owner"], "inline": True},
|
||
{"name": "Tier", "value": ctx_data["tier"], "inline": True},
|
||
{"name": "Duration", "value": ctx_data["duration"], "inline": True},
|
||
{"name": "Timestamp", "value": ctx_data["timestamp"], "inline": True}
|
||
],
|
||
"url": ctx_data["log_url"]
|
||
}
|
||
]
|
||
}
|
||
results["discord"] = cls.send_discord_alert(discord_url, discord_payload)
|
||
|
||
# 4. SITOMAT Gateway Incident Dispatch
|
||
sitomat_url = os.getenv("ALERT_SITOMAT_GATEWAY_URL")
|
||
if alert_config and "channels" in alert_config:
|
||
sitomat_conf = alert_config["channels"].get("sitomat", {})
|
||
if sitomat_conf.get("enabled", False):
|
||
sitomat_url = sitomat_conf.get("endpoint_url", sitomat_url)
|
||
|
||
if sitomat_url:
|
||
sitomat_incident_payload = {
|
||
"incidentType": incident_type,
|
||
"pipelineId": ctx_data["dag_id"],
|
||
"taskId": ctx_data["task_id"],
|
||
"severity": ctx_data["severity"],
|
||
"errorDetails": ctx_data["error_msg"],
|
||
"timestamp": ctx_data["timestamp"],
|
||
"governance": {
|
||
"domain": ctx_data["domain"],
|
||
"owner": ctx_data["owner"],
|
||
"tier": ctx_data["tier"]
|
||
}
|
||
}
|
||
results["sitomat"] = cls.send_sitomat_alert(sitomat_url, sitomat_incident_payload)
|
||
|
||
return results
|
||
|
||
# =========================================================================
|
||
# Airflow Standard Lifecycle Callbacks
|
||
# =========================================================================
|
||
|
||
@classmethod
|
||
def on_failure_callback(cls, context: Dict[str, Any], alert_config: Optional[Dict[str, Any]] = None):
|
||
"""
|
||
Callback resmi saat Task atau DAG mengalami kegagalan eksekusi (Exit Code != 0 / Exception).
|
||
"""
|
||
try:
|
||
ctx_data = cls.extract_context(context)
|
||
logger.info("Executing MericaAlertHook.on_failure_callback for %s / %s", ctx_data["dag_id"], ctx_data["task_id"])
|
||
return cls.dispatch_alert("TASK_FAILURE", ctx_data, alert_config)
|
||
except Exception as e:
|
||
logger.error("Error inside MericaAlertHook.on_failure_callback: %s", e)
|
||
return {"error": str(e)}
|
||
|
||
@classmethod
|
||
def sla_miss_callback(
|
||
cls,
|
||
dag: Any,
|
||
task_list: Any,
|
||
blocking_task_list: Any,
|
||
slas: Any,
|
||
blocking_tis: Any,
|
||
alert_config: Optional[Dict[str, Any]] = None
|
||
):
|
||
"""
|
||
Callback resmi Airflow saat pipeline berjalan melampaui toleransi SLA.
|
||
"""
|
||
try:
|
||
dag_id = getattr(dag, "dag_id", "unknown_pipeline")
|
||
logger.warning("Executing MericaAlertHook.sla_miss_callback for %s", dag_id)
|
||
|
||
ctx_data = {
|
||
"dag_id": dag_id,
|
||
"task_id": "sla_miss_watcher",
|
||
"try_number": 1,
|
||
"max_tries": 1,
|
||
"duration": "OVERRUN",
|
||
"log_url": f"http://localhost:8088/dags/{dag_id}/grid",
|
||
"error_msg": f"Pipeline execution exceeded SLA threshold. Blocking tasks: {blocking_task_list}",
|
||
"domain": "Lingkungan Hidup",
|
||
"owner": "dlh_data_engineering_group",
|
||
"tier": "Tier.Tier1",
|
||
"severity": "⏱️ P1 (SLA MISS)",
|
||
"color_hex": "#FFA500",
|
||
"timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
|
||
}
|
||
return cls.dispatch_alert("SLA_MISS", ctx_data, alert_config)
|
||
except Exception as e:
|
||
logger.error("Error inside MericaAlertHook.sla_miss_callback: %s", e)
|
||
return {"error": str(e)}
|
||
|
||
@classmethod
|
||
def on_quality_gate_breach(
|
||
cls,
|
||
context: Dict[str, Any],
|
||
failed_expectations: List[Dict[str, Any]],
|
||
success_rate: float,
|
||
alert_config: Optional[Dict[str, Any]] = None
|
||
):
|
||
"""
|
||
Callback khusus saat Great Expectations mengalami breach kualitas data.
|
||
"""
|
||
try:
|
||
ctx_data = cls.extract_context(context)
|
||
ctx_data["severity"] = "🛡️ P1 (QUALITY GATE BREACH)"
|
||
ctx_data["color_hex"] = "#E01E5A"
|
||
|
||
rules_summary = "\n".join([
|
||
f"- Rule: {f.get('rule', 'unknown')} | Column: {f.get('column', 'table')}"
|
||
for f in failed_expectations[:5]
|
||
])
|
||
score_str = f"{success_rate:.1f}%" if isinstance(success_rate, (int, float)) else f"{success_rate}%"
|
||
ctx_data["error_msg"] = (
|
||
f"Quality Score: {score_str}\n"
|
||
f"Failed Rules ({len(failed_expectations)} total):\n{rules_summary}"
|
||
)
|
||
|
||
logger.warning("Executing MericaAlertHook.on_quality_gate_breach for %s", ctx_data["dag_id"])
|
||
return cls.dispatch_alert("QUALITY_GATE_BREACH", ctx_data, alert_config)
|
||
except Exception as e:
|
||
logger.error("Error inside MericaAlertHook.on_quality_gate_breach: %s", e)
|
||
return {"error": str(e)}
|