diff --git a/configs/e2e_orders_master_medallion_flow.yaml b/configs/e2e_orders_master_medallion_flow.yaml index 6ef0e44..30dbb79 100644 --- a/configs/e2e_orders_master_medallion_flow.yaml +++ b/configs/e2e_orders_master_medallion_flow.yaml @@ -2,6 +2,12 @@ pipeline_id: e2e_orders_master_medallion_flow description: Dynamic Data Lakehouse Pipeline via DLH Studio schedule: 0 */1 * * * mode: flexible +governance: + domain: Lingkungan Hidup & Kehutanan + owner: de_team_dlh + owner_name: Data Engineering DLH + tier: Tier.Tier1 + certification: Certification.Gold tasks: - task_id: node-9826 type: source_db diff --git a/dags/dag_factory.py b/dags/dag_factory.py index 40a27bd..1bfabdd 100644 --- a/dags/dag_factory.py +++ b/dags/dag_factory.py @@ -42,11 +42,19 @@ def notify_openmetadata_status(dag_id: str, execution_status: str, task_names: l token = None om_hosts = [os.getenv("OPENMETADATA_HOST", "http://openmetadata:8585"), "http://localhost:8585", "http://127.0.0.1:8585"] active_host = None + om_user = os.getenv("OPENMETADATA_USERNAME", "admin@open-metadata.org") + om_pass = os.getenv("OPENMETADATA_PASSWORD", "admin") + if om_pass == "admin": + b64_pass = "YWRtaW4=" + else: + import base64 + b64_pass = base64.b64encode(om_pass.encode("utf-8")).decode("utf-8") + for h in om_hosts: try: login_req = urllib.request.Request( f"{h}/api/v1/users/login", - data=json.dumps({"email": "admin@open-metadata.org", "password": "YWRtaW4="}).encode("utf-8"), + data=json.dumps({"email": om_user, "password": b64_pass}).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST" ) @@ -60,11 +68,15 @@ def notify_openmetadata_status(dag_id: str, execution_status: str, task_names: l if not token or not active_host: return - fqn = f"dlh_airflow_service.{dag_id}" + pipeline_svc = os.getenv("OPENMETADATA_PIPELINE_SERVICE_NAME", "dlh_airflow_service") + airflow_host = os.getenv("OPENMETADATA_AIRFLOW_HOST", "http://merica-airflow-webserver:8080") + fqn = f"{pipeline_svc}.{dag_id}" if not task_names: task_names = ["bronze_ingestion_task", "silver_transformation_task", "gold_data_mart_task"] now_ms = int(time.time() * 1000) + dag = context.get('dag') if context else None + dag_params = getattr(dag, 'params', {}) if dag else {} dag_run = context.get('dag_run') if context else None dag_start = int(dag_run.start_date.timestamp() * 1000) if (dag_run and getattr(dag_run, 'start_date', None)) else (now_ms - 60000) dag_end = int(dag_run.end_date.timestamp() * 1000) if (dag_run and getattr(dag_run, 'end_date', None)) else now_ms @@ -111,16 +123,23 @@ def notify_openmetadata_status(dag_id: str, execution_status: str, task_names: l "taskStatus": task_statuses } - # Auto-provision pipelineService if not already created + # Dynamic Governance Resolution (Zero Hardcoding: Context > Env > Defaults) + gov_domain = (dag_params.get('domain') if isinstance(dag_params, dict) else None) or os.getenv("GOVERNANCE_DOMAIN", "Lingkungan Hidup & Kehutanan") + gov_owner_team = (dag_params.get('owner') if isinstance(dag_params, dict) else None) or os.getenv("GOVERNANCE_OWNER_TEAM", "de_team_dlh") + gov_owner_name = os.getenv("GOVERNANCE_OWNER_NAME", "Data Engineering DLH") + gov_tier = (dag_params.get('tier') if isinstance(dag_params, dict) else None) or os.getenv("GOVERNANCE_TIER", "Tier.Tier1") + gov_cert = (dag_params.get('certification') if isinstance(dag_params, dict) else None) or os.getenv("GOVERNANCE_CERTIFICATION", "Certification.Gold") + + # 1. Auto-provision pipelineService if not already created try: svc_payload = { - "name": "dlh_airflow_service", - "displayName": "DLH Airflow Service", + "name": pipeline_svc, + "displayName": pipeline_svc.replace("_", " ").title(), "serviceType": "Airflow", "connection": { "config": { "type": "Airflow", - "hostPort": "http://merica-airflow-webserver:8080" + "hostPort": airflow_host } } } @@ -135,25 +154,146 @@ def notify_openmetadata_status(dag_id: str, execution_status: str, task_names: l except Exception: pass - # Auto-provision pipeline entity in OpenMetadata + # 2. Auto-provision Domain in OpenMetadata if missing + domain_id = None + domain_name = None + if gov_domain: + try: + dom_slug = gov_domain.replace(" ", "_").replace("&", "").replace("__", "_").strip("_") + try: + d_req = urllib.request.Request( + f"{active_host}/api/v1/domains/name/{dom_slug}", + headers={"Authorization": f"Bearer {token}"} + ) + with urllib.request.urlopen(d_req, timeout=3) as d_res: + d_data = json.loads(d_res.read().decode("utf-8")) + domain_id = d_data.get("id") + domain_name = d_data.get("name") + except urllib.error.HTTPError as he: + if he.code == 404: + dom_payload = { + "name": dom_slug, + "displayName": gov_domain, + "domainType": "Aggregate", + "description": f"Domain {gov_domain} managed by MERICA" + } + c_req = urllib.request.Request( + f"{active_host}/api/v1/domains", + data=json.dumps(dom_payload).encode("utf-8"), + headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, + method="POST" + ) + with urllib.request.urlopen(c_req, timeout=3) as c_res: + c_data = json.loads(c_res.read().decode("utf-8")) + domain_id = c_data.get("id") + domain_name = c_data.get("name") + except Exception as e: + print(f"Domain provisioning note: {e}") + + # 3. Auto-provision Team in OpenMetadata if missing + owner_id = None + if gov_owner_team: + try: + team_slug = gov_owner_team.replace(" ", "_").lower() + try: + t_req = urllib.request.Request( + f"{active_host}/api/v1/teams/name/{team_slug}", + headers={"Authorization": f"Bearer {token}"} + ) + with urllib.request.urlopen(t_req, timeout=3) as t_res: + t_data = json.loads(t_res.read().decode("utf-8")) + owner_id = t_data.get("id") + except urllib.error.HTTPError as he: + if he.code == 404: + team_payload = { + "name": team_slug, + "displayName": gov_owner_name or gov_owner_team, + "teamType": "Group", + "description": f"Team {gov_owner_name or gov_owner_team} managed by MERICA" + } + c_req = urllib.request.Request( + f"{active_host}/api/v1/teams", + data=json.dumps(team_payload).encode("utf-8"), + headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, + method="POST" + ) + with urllib.request.urlopen(c_req, timeout=3) as c_res: + c_data = json.loads(c_res.read().decode("utf-8")) + owner_id = c_data.get("id") + except Exception as e: + print(f"Team provisioning note: {e}") + + # 4. Auto-provision pipeline entity in OpenMetadata with Governance attributes + pipe_id = None try: pipe_payload = { "name": dag_id, "displayName": dag_id.replace("_", " ").title(), - "service": "dlh_airflow_service", + "service": pipeline_svc, "description": f"Managed Airflow Pipeline {dag_id}", "tasks": [{"name": str(t), "displayName": str(t), "taskType": "custom_task"} for t in (task_names or [])] } + if domain_name: + pipe_payload["domains"] = [domain_name] + if owner_id: + pipe_payload["owners"] = [{"id": owner_id, "type": "team"}] + if gov_tier: + pipe_payload["tags"] = [{"tagFQN": gov_tier, "source": "Classification", "labelType": "Manual", "state": "Confirmed"}] + pipe_req = urllib.request.Request( f"{active_host}/api/v1/pipelines", data=json.dumps(pipe_payload).encode("utf-8"), headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, method="PUT" ) - with urllib.request.urlopen(pipe_req, timeout=4) as _: - pass - except Exception: - pass + with urllib.request.urlopen(pipe_req, timeout=4) as p_resp: + p_data = json.loads(p_resp.read().decode("utf-8")) + pipe_id = p_data.get("id") + except Exception as e: + print(f"Pipeline PUT note: {e}") + + # 5. Patch Governance attributes (Domains, Owners, Tier, Certification) + try: + if not pipe_id: + fetch_r = urllib.request.Request( + f"{active_host}/api/v1/pipelines/name/{fqn}", + headers={"Authorization": f"Bearer {token}"} + ) + with urllib.request.urlopen(fetch_r, timeout=3) as f_res: + pipe_id = json.loads(f_res.read().decode("utf-8")).get("id") + + if pipe_id: + patch_ops = [] + if domain_id: + patch_ops.append({"op": "add", "path": "/domains", "value": [{"id": domain_id, "type": "domain"}]}) + if owner_id: + patch_ops.append({"op": "add", "path": "/owners", "value": [{"id": owner_id, "type": "team"}]}) + if gov_tier: + patch_ops.append({"op": "add", "path": "/tags", "value": [{"tagFQN": gov_tier, "source": "Classification", "labelType": "Manual", "state": "Confirmed"}]}) + if gov_cert: + patch_ops.append({ + "op": "add", + "path": "/certification", + "value": { + "tagLabel": { + "tagFQN": gov_cert, + "source": "Classification", + "labelType": "Manual", + "state": "Confirmed" + } + } + }) + if patch_ops: + patch_req = urllib.request.Request( + f"{active_host}/api/v1/pipelines/{pipe_id}", + data=json.dumps(patch_ops).encode("utf-8"), + headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json-patch+json"}, + method="PATCH" + ) + with urllib.request.urlopen(patch_req, timeout=4) as _: + pass + except Exception as e: + print(f"Pipeline Governance PATCH note: {e}") req = urllib.request.Request( f"{active_host}/api/v1/pipelines/{fqn}/status", @@ -203,17 +343,23 @@ def build_dag_from_yaml(config_file: str) -> DAG: dag_id = re.sub(r'[^a-zA-Z0-9_.-]', '_', str(raw_id).strip()).lower() dag_id = re.sub(r'_+', '_', dag_id).strip('_') + default_domain = os.getenv('GOVERNANCE_DOMAIN', 'Lingkungan Hidup & Kehutanan') + default_owner = os.getenv('GOVERNANCE_OWNER_TEAM', 'de_team_dlh') + default_tier = os.getenv('GOVERNANCE_TIER', 'Tier.Tier1') + default_certification = os.getenv('GOVERNANCE_CERTIFICATION', 'Certification.Gold') + schedule = cfg.get('schedule', '@daily') - owner = cfg.get('owner', 'de_team_dlh') + owner = cfg.get('owner') or default_owner tags = cfg.get('tags', ['dlh', 'medallion']) mode = cfg.get('mode', 'medallion') governance = cfg.get('governance', {}) alert_cfg = cfg.get('alerting', {}) dag_params = { - 'domain': governance.get('domain', cfg.get('domain', 'Lingkungan Hidup & Kehutanan')), + 'domain': governance.get('domain', cfg.get('domain', default_domain)), 'owner': governance.get('owner', owner), - 'tier': governance.get('tier', cfg.get('tier', 'Tier.Tier2')), + 'tier': governance.get('tier', cfg.get('tier', default_tier)), + 'certification': governance.get('certification', cfg.get('certification', default_certification)), 'alerting': alert_cfg } diff --git a/dags/helpers/base_lakehouse_operator.py b/dags/helpers/base_lakehouse_operator.py index 5e90a1e..c5cc9e8 100644 --- a/dags/helpers/base_lakehouse_operator.py +++ b/dags/helpers/base_lakehouse_operator.py @@ -93,7 +93,8 @@ class BaseLakehouseOperator(BaseOperator): logger.debug("OpenMetadata host/token unavailable; skipping lineage emission.") return - pipeline_fqn = f"dlh_airflow_service.{dag_id}" + pipeline_svc = os.getenv("OPENMETADATA_PIPELINE_SERVICE_NAME", "dlh_airflow_service") + pipeline_fqn = f"{pipeline_svc}.{dag_id}" pipeline_id_uuid = self._resolve_entity_uuid(active_host, token, "pipeline", pipeline_fqn) # Emit lineage for each inlet -> outlet pair @@ -141,13 +142,16 @@ class BaseLakehouseOperator(BaseOperator): e.g. 's3://dlh-bronze/db_cdc/environmental_permits' -> 'dlh_s3_storage.dlh-bronze.db_cdc.environmental_permits' e.g. 'silver.environmental_permits_clean' -> 'dlh_postgres_service.dlh_lakehouse.silver.environmental_permits_clean' """ + storage_svc = os.getenv("OPENMETADATA_STORAGE_SERVICE_NAME", "dlh_minio_s3") + db_svc = os.getenv("OPENMETADATA_DB_SERVICE_NAME", "dlh_postgres_service") + db_name = os.getenv("POSTGRES_DB", "dlh_lakehouse") clean = dataset_str.strip() if clean.startswith("s3://"): parts = clean.replace("s3://", "").strip("/").split("/") - return f"dlh_minio_s3.{'.'.join(parts)}" + return f"{storage_svc}.{'.'.join(parts)}" elif "." in clean: - return f"dlh_postgres_service.dlh_lakehouse.{clean}" - return f"dlh_postgres_service.dlh_lakehouse.public.{clean}" + return f"{db_svc}.{db_name}.{clean}" + return f"{db_svc}.{db_name}.public.{clean}" def _get_openmetadata_auth(self) -> tuple: """Authenticate and get access token.""" @@ -156,11 +160,19 @@ class BaseLakehouseOperator(BaseOperator): "http://localhost:8585", "http://127.0.0.1:8585" ] + om_user = os.getenv("OPENMETADATA_USERNAME", "admin@open-metadata.org") + om_pass = os.getenv("OPENMETADATA_PASSWORD", "admin") + if om_pass == "admin": + b64_pass = "YWRtaW4=" + else: + import base64 + b64_pass = base64.b64encode(om_pass.encode("utf-8")).decode("utf-8") + for h in hosts: try: login_req = urllib.request.Request( f"{h}/api/v1/users/login", - data=json.dumps({"email": "admin@open-metadata.org", "password": "YWRtaW4="}).encode("utf-8"), + data=json.dumps({"email": om_user, "password": b64_pass}).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST" ) diff --git a/dags/helpers/ge_lakehouse_operator.py b/dags/helpers/ge_lakehouse_operator.py index d7375b2..9c51ebb 100644 --- a/dags/helpers/ge_lakehouse_operator.py +++ b/dags/helpers/ge_lakehouse_operator.py @@ -352,9 +352,17 @@ class GreatExpectationsLakehouseOperator(BaseLakehouseOperator): try: # Login to get token + om_user = os.getenv("OPENMETADATA_USERNAME", "admin@open-metadata.org") + om_pass = os.getenv("OPENMETADATA_PASSWORD", "admin") + if om_pass == "admin": + b64_pass = "YWRtaW4=" + else: + import base64 + b64_pass = base64.b64encode(om_pass.encode("utf-8")).decode("utf-8") + login_req = urllib.request.Request( f"{om_host}/api/v1/users/login", - data=json.dumps({"email": "admin@open-metadata.org", "password": "YWRtaW4="}).encode("utf-8"), + data=json.dumps({"email": om_user, "password": b64_pass}).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST" ) @@ -364,7 +372,8 @@ class GreatExpectationsLakehouseOperator(BaseLakehouseOperator): if not token: return - fqn = f"dlh_airflow_service.{self.dag_id or 'lakehouse_pipeline'}.{self.task_id}" + pipeline_svc = os.getenv("OPENMETADATA_PIPELINE_SERVICE_NAME", "dlh_airflow_service") + fqn = f"{pipeline_svc}.{self.dag_id or 'lakehouse_pipeline'}.{self.task_id}" payload = { "timestamp": int(time.time() * 1000), "testCaseStatus": "Success" if summary["overall_success"] else "Failed", diff --git a/dags/helpers/merica_alert_hook.py b/dags/helpers/merica_alert_hook.py index 1bd50ad..08fefc6 100644 --- a/dags/helpers/merica_alert_hook.py +++ b/dags/helpers/merica_alert_hook.py @@ -293,6 +293,9 @@ class MericaAlertHook: 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") @@ -313,7 +316,7 @@ class MericaAlertHook: f"👤 Owner: {ctx_data['owner']} | Tier: {ctx_data['tier']}\n\n" f"⚠️ Root Cause:\n{ctx_data['error_msg']}\n\n" f"🔗 Airflow Log | " - f"OpenMetadata Lineage" + f"OpenMetadata Lineage" ) results["telegram"] = cls.send_telegram_alert(tg_bot_token, tg_chat_id, tg_message) @@ -354,7 +357,7 @@ class MericaAlertHook: "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"http://localhost:8585/pipeline/dlh_airflow_service.{ctx_data['dag_id']}/lineage"} + {"type": "button", "text": {"type": "plain_text", "text": "OpenMetadata Lineage"}, "url": f"{om_public_url}/pipeline/{pipeline_svc}.{ctx_data['dag_id']}/lineage"} ] } ]