2026-10-08 19:33:15 +07:00
|
|
|
|
"""
|
|
|
|
|
|
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] = {}
|
|
|
|
|
|
|
2026-10-19 02:40:31 +07:00
|
|
|
|
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")
|
|
|
|
|
|
|
2026-10-08 19:33:15 +07:00
|
|
|
|
# 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> | "
|
2026-10-19 02:40:31 +07:00
|
|
|
|
f"<a href='{om_public_url}/pipeline/{pipeline_svc}.{ctx_data['dag_id']}/lineage'>OpenMetadata Lineage</a>"
|
2026-10-08 19:33:15 +07:00
|
|
|
|
)
|
|
|
|
|
|
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"]},
|
2026-10-19 02:40:31 +07:00
|
|
|
|
{"type": "button", "text": {"type": "plain_text", "text": "OpenMetadata Lineage"}, "url": f"{om_public_url}/pipeline/{pipeline_svc}.{ctx_data['dag_id']}/lineage"}
|
2026-10-08 19:33:15 +07:00
|
|
|
|
]
|
|
|
|
|
|
}
|
|
|
|
|
|
]
|
|
|
|
|
|
}
|
|
|
|
|
|
]
|
|
|
|
|
|
}
|
|
|
|
|
|
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)}
|
