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
|