217 lines
8.3 KiB
Python
217 lines
8.3 KiB
Python
"""
|
||
Sitomat Callback Operator for MERICA Data Lakehouse Studio
|
||
Mengirimkan status penyelesaian layer Medallion (Bronze/Silver/Gold) ke API Sitomat F16 Ingestion Hub.
|
||
Mendukung idempotensi, penanganan konflik (HTTP 409), dan fallback mock mode untuk pengembangan lokal.
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import logging
|
||
from typing import Any, Dict, Optional, Sequence
|
||
import requests
|
||
from requests.adapters import HTTPAdapter
|
||
from urllib3.util.retry import Retry
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# BaseOperator import with fallback for testing outside Airflow
|
||
try:
|
||
from airflow.models.baseoperator import BaseOperator
|
||
except ImportError:
|
||
class BaseOperator:
|
||
def __init__(self, *args, **kwargs):
|
||
self.task_id = kwargs.get("task_id", "sitomat_callback_task")
|
||
self.log = logging.getLogger(self.__class__.__name__)
|
||
|
||
|
||
def create_sitomat_session(
|
||
max_retries: int = 3,
|
||
backoff_factor: float = 1.0,
|
||
status_forcelist = (500, 502, 503, 504)
|
||
) -> requests.Session:
|
||
"""Mengonfigurasi HTTP session dengan connection pool dan retry strategy."""
|
||
session = requests.Session()
|
||
retry_strategy = Retry(
|
||
total=max_retries,
|
||
backoff_factor=backoff_factor,
|
||
status_forcelist=status_forcelist,
|
||
allowed_methods=["POST", "PUT"]
|
||
)
|
||
adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=5, pool_maxsize=5)
|
||
session.mount("http://", adapter)
|
||
session.mount("https://", adapter)
|
||
return session
|
||
|
||
|
||
class SitomatCallbackOperator(BaseOperator):
|
||
"""
|
||
Operator untuk memanggil webhook callback Sitomat F16 Ingestion Hub:
|
||
POST /ingest/v1/runs/:id/{bronze,silver,gold}
|
||
|
||
:param run_id: ID Ingestion Run yang diinisiasi oleh Sitomat (e.g. 'run-20261005-001')
|
||
:param layer: Layer yang dilaporkan ('bronze' | 'silver' | 'gold')
|
||
:param artifact_ref: Referensi tabel/berkas (e.g. 'iceberg.silver.orders_silver')
|
||
:param checksum: Hash snapshot ID, dbt invocation ID, atau file checksum
|
||
:param derived_from_version_id: ID versi sumber (wajib untuk layer silver)
|
||
:param sitomat_base_url: Base URL API Sitomat (Default: env SITOMAT_API_URL atau 'http://localhost:3000')
|
||
:param api_key: API Key ber-scope 'pipeline' (Default: env SITOMAT_PIPELINE_API_KEY)
|
||
:param metadata_payload: Kamus data tambahan (GE metrics, record count, openmetadata_fqn)
|
||
:param mock_mode: Jika True atau env SITOMAT_MOCK_MODE='true', log dan return sukses tanpa call fisik
|
||
"""
|
||
|
||
template_fields: Sequence[str] = (
|
||
"run_id",
|
||
"artifact_ref",
|
||
"checksum",
|
||
"derived_from_version_id",
|
||
"metadata_payload"
|
||
)
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
run_id: Optional[str] = "auto",
|
||
layer: str,
|
||
artifact_ref: str,
|
||
checksum: str = "sha256-default",
|
||
derived_from_version_id: Optional[str] = None,
|
||
sitomat_base_url: Optional[str] = None,
|
||
api_key: Optional[str] = None,
|
||
metadata_payload: Optional[Dict[str, Any]] = None,
|
||
mock_mode: Optional[bool] = None,
|
||
**kwargs
|
||
):
|
||
super().__init__(**kwargs)
|
||
self.run_id = str(run_id or "auto")
|
||
self.layer = layer.lower().strip()
|
||
self.artifact_ref = artifact_ref
|
||
self.checksum = checksum
|
||
self.derived_from_version_id = derived_from_version_id
|
||
|
||
self.sitomat_base_url = (
|
||
sitomat_base_url
|
||
or os.getenv("SITOMAT_API_URL", "http://localhost:3000")
|
||
).rstrip("/")
|
||
self.api_key = api_key or os.getenv("SITOMAT_PIPELINE_API_KEY", "")
|
||
self.metadata_payload = metadata_payload or {}
|
||
|
||
if mock_mode is not None:
|
||
self.mock_mode = mock_mode
|
||
else:
|
||
self.mock_mode = os.getenv("SITOMAT_MOCK_MODE", "false").lower() in ("true", "1", "yes")
|
||
|
||
if self.layer not in ("bronze", "silver", "gold"):
|
||
raise ValueError(f"Layer '{self.layer}' tidak valid. Pilihan: bronze, silver, gold.")
|
||
|
||
def execute(self, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""
|
||
Mengeksekusi HTTP POST callback ke endpoint Sitomat F16.
|
||
"""
|
||
import time
|
||
effective_run_id = self.run_id
|
||
if effective_run_id in ("auto", "default", "") or not effective_run_id:
|
||
if context:
|
||
if "run_id" in context:
|
||
effective_run_id = str(context["run_id"])
|
||
elif "dag_run" in context and hasattr(context["dag_run"], "run_id"):
|
||
effective_run_id = str(context["dag_run"].run_id)
|
||
elif "ti" in context and hasattr(context["ti"], "run_id"):
|
||
effective_run_id = str(context["ti"].run_id)
|
||
if not effective_run_id or effective_run_id in ("auto", "default", ""):
|
||
effective_run_id = f"run-{int(time.time())}"
|
||
|
||
endpoint = f"{self.sitomat_base_url}/ingest/v1/runs/{effective_run_id}/{self.layer}"
|
||
payload: Dict[str, Any] = {
|
||
"artifactRef": self.artifact_ref,
|
||
"checksum": self.checksum,
|
||
"metadata": self.metadata_payload,
|
||
}
|
||
|
||
if self.derived_from_version_id:
|
||
payload["derivedFromVersionId"] = self.derived_from_version_id
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
}
|
||
if self.api_key:
|
||
headers["Authorization"] = f"Bearer {self.api_key}"
|
||
|
||
logger.info(
|
||
"Mengirimkan callback Sitomat: layer=%s, run_id=%s, endpoint=%s",
|
||
self.layer, self.run_id, endpoint
|
||
)
|
||
|
||
# Mock Mode Guardrail (untuk testing lokal saat server Sitomat belum aktif)
|
||
if self.mock_mode:
|
||
logger.info("ℹ️ SITOMAT_MOCK_MODE aktif: Melompati panggilan HTTP riil.")
|
||
mock_response = {
|
||
"status": "mock_success",
|
||
"layer": self.layer,
|
||
"runId": self.run_id,
|
||
"artifactRef": self.artifact_ref,
|
||
"acknowledged": True,
|
||
"note": "Callback diakui oleh mock mode MERICA"
|
||
}
|
||
return mock_response
|
||
|
||
session = create_sitomat_session()
|
||
try:
|
||
response = session.post(
|
||
url=endpoint,
|
||
json=payload,
|
||
headers=headers,
|
||
timeout=(5.0, 15.0) # (connect timeout, read timeout)
|
||
)
|
||
|
||
# 1. Sukses Baru (200 OK / 201 Created)
|
||
if response.status_code in (200, 201):
|
||
try:
|
||
res_json = response.json()
|
||
except Exception:
|
||
res_json = {"raw": response.text}
|
||
logger.info("✅ Callback Sitomat berhasil (HTTP %d): %s", response.status_code, res_json)
|
||
return {
|
||
"status": "success",
|
||
"statusCode": response.status_code,
|
||
"response": res_json
|
||
}
|
||
|
||
# 2. Idempoten Replay / Konflik Terdeteksi (HTTP 409)
|
||
elif response.status_code == 409:
|
||
logger.warning(
|
||
"⚠️ Idempoten/Conflict terdeteksi pada Sitomat (HTTP 409): %s",
|
||
response.text
|
||
)
|
||
return {
|
||
"status": "conflict_acknowledged",
|
||
"statusCode": 409,
|
||
"response": response.text,
|
||
"note": "Replay run callback diakui secara idempoten"
|
||
}
|
||
|
||
# 3. Kesalahan Lain (4xx / 5xx)
|
||
else:
|
||
err_msg = (
|
||
f"Callback Sitomat gagal dengan status {response.status_code} "
|
||
f"pada {endpoint}: {response.text}"
|
||
)
|
||
logger.error(err_msg)
|
||
raise RuntimeError(err_msg)
|
||
|
||
except requests.exceptions.ConnectionError as conn_err:
|
||
logger.warning(
|
||
"⚠️ Tidak dapat menghubungi server Sitomat di %s: %s. "
|
||
"Memperlakukan sebagai warning berkat mode toleran koneksi.",
|
||
endpoint, conn_err
|
||
)
|
||
return {
|
||
"status": "connection_warning",
|
||
"endpoint": endpoint,
|
||
"error": str(conn_err),
|
||
"acknowledged": False
|
||
}
|
||
except Exception as e:
|
||
logger.error("Terjadi pengecualian saat memanggil callback Sitomat: %s", e)
|
||
raise
|