feat(openmetadata): backfill Airflow task detail + dataset lineage

OM 1.13 cannot deserialize Airflow 3.0.6 serialized DAGs, so pipelines had no
tasks. af_tasks.py parses serialized_dag and PATCHes task lists; af_lineage.py
adds generator->table and mask_to_curated/hadoop_to_trino lineage edges.

Note: OM MySQL datadir lives on the 448MB /opt partition which had filled up
(103MB binlog). Disabled binary logging + shrank redo log capacity in the OM
docker-compose to restore headroom.

deploy/openmetadata/ingest/af_lineage.py

@@ -0,0 +1,85 @@
+"""Add dataset lineage for the Airflow DAGs into OpenMetadata.
+
+- Generator DAGs produce a source table  -> pipeline:table edge
+- ETL DAGs transform source -> target     -> table:table edge carrying the pipeline
+"""
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+
+OM = "http://openmetadata-server:8585/api"
+TOKEN = os.environ["OM_TOKEN"]
+HDRS = {"Authorization": "Bearer " + TOKEN}
+
+
+def req(method, path, body=None):
+    data = json.dumps(body).encode() if body is not None else None
+    h = dict(HDRS)
+    if data is not None:
+        h["Content-Type"] = "application/json"
+    last = "?"
+    for _ in range(3):
+        r = urllib.request.Request(OM + path, data=data, headers=h, method=method)
+        try:
+            with urllib.request.urlopen(r, timeout=60) as resp:
+                return resp.status, json.loads(resp.read().decode() or "{}")
+        except urllib.error.HTTPError as e:
+            return e.code, e.read().decode()
+        except Exception as e:  # noqa: BLE001 - timeouts / transient
+            last = repr(e)
+    return 0, last
+
+
+def get_id(kind, fqn):
+    base = "/v1/pipelines/name/" if kind == "pipeline" else "/v1/tables/name/"
+    st, d = req("GET", base + urllib.parse.quote(fqn, safe=""))
+    if st == 200:
+        return d["id"]
+    print("  ! cannot resolve", kind, fqn, "->", st)
+    return None
+
+
+def edge(from_kind, from_fqn, to_fqn, pipeline_fqn=None):
+    fid = get_id(from_kind, from_fqn)
+    tid = get_id("table", to_fqn)
+    if not fid or not tid:
+        return
+    body = {
+        "edge": {
+            "fromEntity": {"id": fid, "type": from_kind},
+            "toEntity": {"id": tid, "type": "table"},
+        }
+    }
+    if pipeline_fqn:
+        pid = get_id("pipeline", pipeline_fqn)
+        if pid:
+            body["edge"]["lineageDetails"] = {
+                "pipeline": {"id": pid, "type": "pipeline"}
+            }
+    st, res = req("PUT", "/v1/lineage", body)
+    label = (pipeline_fqn or from_fqn).split(".")[-1]
+    print("  %-22s %s -> %s" % (label, from_fqn.split(".")[-1], to_fqn.split(".")[-1]),
+          "OK" if st in (200, 201) else ("ERR %s %s" % (st, res)), flush=True)
+
+
+PG = "atc_postgres.postgres.public.sales_orders"
+MY = "atc_mysql.default.hr.employee_events"
+MO = "atc_mongodb.supplychain.supplychain.events"
+
+print("generators:")
+edge("pipeline", "atc_airflow.gen_postgres", PG)
+edge("pipeline", "atc_airflow.gen_mysql", MY)
+edge("pipeline", "atc_airflow.gen_mongodb", MO)
+edge("pipeline", "atc_airflow.generate_data_all_databases", PG)
+edge("pipeline", "atc_airflow.generate_data_all_databases", MY)
+edge("pipeline", "atc_airflow.generate_data_all_databases", MO)
+
+print("ETL (table -> table via pipeline):")
+edge("table", PG, "atc_trino.iceberg.curated_masked.sales_orders_masked",
+     "atc_airflow.mask_to_curated")
+edge("table", MY, "atc_trino.iceberg.curated_masked.employee_events_masked",
+     "atc_airflow.mask_to_curated")
+edge("pipeline", "atc_airflow.hadoop_to_trino",
+     "atc_trino.iceberg.hadoop.historical_sales_hdfs")

deploy/openmetadata/ingest/af_tasks.py

@@ -0,0 +1,92 @@
+"""Backfill task-level detail into OpenMetadata Airflow pipelines.
+
+OM 1.13's Airflow connector cannot deserialize Airflow 3.0.6 serialized DAGs, so
+pipelines land with zero tasks. We parse the serialized_dag table ourselves and
+PATCH each pipeline with its tasks + intra-DAG dependencies (downstreamTasks).
+"""
+import json
+import os
+import sqlite3
+import urllib.error
+import urllib.request
+import zlib
+
+DB = "/" + "/".join(["tmp", "airflow.db"])
+OM = "http://openmetadata-server:8585/api"
+TOKEN = os.environ["OM_TOKEN"]
+SERVICE = "atc_airflow"
+
+HDRS = {"Authorization": "Bearer " + TOKEN}
+
+
+def req(method, path, body=None, ctype="application/json"):
+    data = json.dumps(body).encode() if body is not None else None
+    h = dict(HDRS)
+    if data is not None:
+        h["Content-Type"] = ctype
+    r = urllib.request.Request(OM + path, data=data, headers=h, method=method)
+    try:
+        with urllib.request.urlopen(r, timeout=30) as resp:
+            return resp.status, json.loads(resp.read().decode() or "{}")
+    except urllib.error.HTTPError as e:
+        return e.code, e.read().decode()
+
+
+def latest_serialized():
+    c = sqlite3.connect(DB)
+    rows = c.execute(
+        "SELECT s.dag_id, s.data, s.data_compressed FROM serialized_dag s "
+        "JOIN (SELECT dag_id, MAX(id) mid FROM serialized_dag GROUP BY dag_id) m "
+        "ON s.id = m.mid"
+    ).fetchall()
+    out = {}
+    for dag_id, data, comp in rows:
+        if data is None and comp is not None:
+            data = zlib.decompress(comp).decode()
+        d = json.loads(data) if isinstance(data, str) else data
+        out[dag_id] = d.get("dag", d)
+    return out
+
+
+def build_tasks(dag):
+    tasks = []
+    for t in dag.get("tasks", []):
+        tt = t.get("__var", t) if isinstance(t, dict) else t
+        tid = tt.get("task_id")
+        if not tid:
+            continue
+        tasks.append(
+            {
+                "name": tid,
+                "displayName": tid,
+                "taskType": tt.get("task_type") or tt.get("_task_type") or "Task",
+                "downstreamTasks": list(tt.get("downstream_task_ids") or []),
+            }
+        )
+    return tasks
+
+
+def main():
+    dags = latest_serialized()
+    for dag_id, dag in sorted(dags.items()):
+        tasks = build_tasks(dag)
+        fqn = SERVICE + "." + dag_id
+        st, pipe = req("GET", "/v1/pipelines/name/" + fqn)
+        if st != 200:
+            print("SKIP", dag_id, "get-failed", st, pipe)
+            continue
+        pid = pipe["id"]
+        op = "replace" if "tasks" in pipe else "add"
+        patch = [{"op": op, "path": "/tasks", "value": tasks}]
+        st2, res = req(
+            "PATCH", "/v1/pipelines/" + pid, patch, "application/json-patch+json"
+        )
+        edges = sum(len(t["downstreamTasks"]) for t in tasks)
+        print(
+            "%-30s tasks=%-2d edges=%-2d -> %s"
+            % (dag_id, len(tasks), edges, "OK" if st2 == 200 else ("ERR %s %s" % (st2, res)))
+        )
+
+
+if __name__ == "__main__":
+    main()