From 3b247fa2bdf4107b1151bcfacb283b1bb46fb62c Mon Sep 17 00:00:00 2001 From: mo Date: Sat, 27 Jun 2026 02:40:27 +0200 Subject: [PATCH] =?UTF-8?q?feat(om):=20OpenMetadata=20integration=20?= =?UTF-8?q?=E2=80=94=20registry=20node,=20link,=20PII=20tag=20source?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add OpenMetadata node (atc-docker02 .47) to node_registry with UI links. pii_catalog now reads OM column PII tags (Presidio auto-classification) as the authoritative source, merged with the name heuristic; OPENMETADATA_URL wired into the api service (token via atc.env). --- api/node_registry.py | 23 ++++++++++++- api/pii_catalog.py | 82 ++++++++++++++++++++++++++++++++++++-------- docker-compose.yml | 1 + 3 files changed, 91 insertions(+), 15 deletions(-) diff --git a/api/node_registry.py b/api/node_registry.py index d0c33de..0f057f2 100644 --- a/api/node_registry.py +++ b/api/node_registry.py @@ -6,7 +6,7 @@ from typing import Any NODE_IDS = [ "airflow", "db", "debezium", "kafka", "lakehouse", "s3", - "docker", "hadoop", "gpu", "command", + "docker", "hadoop", "gpu", "command", "openmetadata", "mo-commander", "bart-commander", "network-watcher", "mcp-coordinator", ] @@ -21,6 +21,7 @@ NODE_AGENT = { "hadoop": "hadoop-ranger", "gpu": "infra-sentinel", "command": "infra-sentinel", + "openmetadata": "lakehouse-ops", } NODE_REGISTRY: dict[str, dict[str, Any]] = { @@ -200,6 +201,26 @@ NODE_REGISTRY: dict[str, dict[str, Any]] = { ], "commands": ["gpu metrics", "model status", "vram usage"], }, + "openmetadata": { + "label": "OpenMetadata", + "vm": "atc-docker02", + "vmid": 0, + "pve": "pve01", + "ip": "10.0.21.47", + "ssh": "ssh root@10.0.21.47", + "role": "governance", + "color": "#7147e8", + "description": "OpenMetadata 1.13 data catalog & lineage. Ingests PostgreSQL, MySQL, MongoDB and Trino; auto-classifies PII (Presidio NER). Single pane for catalog, lineage and governance across the lab.", + "links": [ + {"label": "OpenMetadata UI", "url": "http://10.0.21.47:8585"}, + {"label": "Ingestion Airflow", "url": "http://10.0.21.47:8080"}, + ], + "endpoints": [ + {"name": "openmetadata", "host": "10.0.21.47", "port": "8585", "proto": "http"}, + {"name": "ingestion", "host": "10.0.21.47", "port": "8080", "proto": "http"}, + ], + "commands": ["catalog status", "run ingestion", "pii classification"], + }, "command": { "label": "Command Center", "vm": "MCP ยท VM304", diff --git a/api/pii_catalog.py b/api/pii_catalog.py index 76cf687..904ba36 100644 --- a/api/pii_catalog.py +++ b/api/pii_catalog.py @@ -25,16 +25,21 @@ OPENMETADATA_URL = os.getenv("OPENMETADATA_URL", "").rstrip("/") OPENMETADATA_TOKEN = os.getenv("OPENMETADATA_TOKEN", "") # Datasets we surface in the PII overlay. node_id matches dataflow.py node ids. +# om_fqn = OpenMetadata table FQN (service.database.schema.table) for tag lookup. DATASETS = [ {"key": "postgres", "node_id": "postgres", "label": "PostgreSQL sales_orders", - "table": "postgres_sales.public.sales_orders", "table_name": "sales_orders", "catalog": "postgres_sales"}, + "table": "postgres_sales.public.sales_orders", "table_name": "sales_orders", "catalog": "postgres_sales", + "om_fqn": "atc_postgres.postgres.public.sales_orders"}, {"key": "mysql", "node_id": "mysql", "label": "MySQL employee_events", - "table": "mysql_hr.hr.employee_events", "table_name": "employee_events", "catalog": "mysql_hr"}, + "table": "mysql_hr.hr.employee_events", "table_name": "employee_events", "catalog": "mysql_hr", + "om_fqn": "atc_mysql.default.hr.employee_events"}, {"key": "mongodb", "node_id": "mongodb", "label": "MongoDB events", - "table": "mongodb_supplychain.supplychain.events", "table_name": "events", "catalog": "mongodb_supplychain"}, + "table": "mongodb_supplychain.supplychain.events", "table_name": "events", "catalog": "mongodb_supplychain", + "om_fqn": "atc_mongodb.default.supplychain.events"}, {"key": "curated", "node_id": "iceberg_curated", "label": "Iceberg curated_masked", "table": "iceberg.curated_masked.sales_orders_masked", "table_name": "sales_orders_masked", "catalog": "iceberg", - "schema": "curated_masked", "masked_layer": True}, + "schema": "curated_masked", "masked_layer": True, + "om_fqn": "atc_trino.iceberg.curated_masked.sales_orders_masked"}, ] # name fragment -> PII category @@ -66,6 +71,37 @@ def _is_masked(col: str, dataset_masked_layer: bool) -> bool: return dataset_masked_layer or c.endswith("_masked") or c.endswith("_hash") or c.endswith("_token") +# OpenMetadata PII tag -> our category. OM applies PII.Sensitive / PII.NonSensitive +# plus optional General/PersonalData tags via auto-classification. +_OM_CAT = { + "PII.Sensitive": "SENSITIVE", + "PII.NonSensitive": "NON_SENSITIVE", +} + + +def _om_column_tags(fqn: str) -> dict[str, list[str]]: + """Return {column_name: [tagFQN,...]} from OpenMetadata for a table FQN.""" + if not OPENMETADATA_URL: + return {} + url = f"{OPENMETADATA_URL}/api/v1/tables/name/{fqn}?fields=columns,tags" + headers = {"Accept": "application/json"} + if OPENMETADATA_TOKEN: + headers["Authorization"] = f"Bearer {OPENMETADATA_TOKEN}" + try: + with httpx.Client(timeout=8.0) as client: + r = client.get(url, headers=headers) + if r.status_code != 200: + return {} + out: dict[str, list[str]] = {} + for c in r.json().get("columns", []) or []: + tags = [t.get("tagFQN") for t in (c.get("tags") or []) if t.get("tagFQN")] + if tags: + out[c["name"]] = tags + return out + except Exception: + return {} + + def _trino_columns(catalog: str, schema: str | None, table_name: str) -> list[str]: sql = ( f"SELECT column_name FROM {catalog}.information_schema.columns " @@ -94,26 +130,44 @@ def _build() -> dict[str, Any]: datasets_out = [] total_pii = 0 total_masked = 0 + om_used = False for ds in DATASETS: cols = _trino_columns(ds["catalog"], ds.get("schema"), ds["table_name"]) + om_tags = _om_column_tags(ds["om_fqn"]) if ds.get("om_fqn") else {} + if om_tags: + om_used = True + # Union of columns known via Trino and via OM (OM may exist before Trino sees it). + all_cols = list(dict.fromkeys(cols + list(om_tags.keys()))) pii_cols = [] - for c in cols: - cat = _classify(c) - if cat: - masked = _is_masked(c, ds.get("masked_layer", False)) - pii_cols.append({"name": c, "category": cat, "masked": masked}) - total_pii += 1 - if masked: - total_masked += 1 + for c in all_cols: + tags = om_tags.get(c, []) + pii_tag = next((t for t in tags if t.startswith("PII.")), None) + heur = _classify(c) + if not pii_tag and not heur: + continue + # Prefer OM PII classification; enrich with heuristic category if present. + if pii_tag: + cat = heur or _OM_CAT.get(pii_tag, "PII") + else: + cat = heur + masked = _is_masked(c, ds.get("masked_layer", False)) + pii_cols.append({ + "name": c, "category": cat, "masked": masked, + "source": "openmetadata" if pii_tag else "heuristic", + "om_tag": pii_tag, + }) + total_pii += 1 + if masked: + total_masked += 1 datasets_out.append({ "key": ds["key"], "node_id": ds["node_id"], "label": ds["label"], "table": ds["table"], - "exists": bool(cols), "masked_layer": ds.get("masked_layer", False), + "exists": bool(all_cols), "masked_layer": ds.get("masked_layer", False), "pii_columns": pii_cols, "pii_count": len(pii_cols), "has_pii": bool(pii_cols), "all_masked": bool(pii_cols) and all(c["masked"] for c in pii_cols), }) return { - "ok": True, "source": "heuristic" if not OPENMETADATA_URL else "openmetadata+heuristic", + "ok": True, "source": "openmetadata+heuristic" if om_used else "heuristic", "datasets": datasets_out, "summary": {"datasets": len(datasets_out), "pii_columns": total_pii, "masked_columns": total_masked, "unmasked_columns": total_pii - total_masked}, diff --git a/docker-compose.yml b/docker-compose.yml index c68d62c..adb5793 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -41,6 +41,7 @@ services: AIRFLOW_URL: http://10.0.21.55:8080 TRINO_URL: http://10.0.21.50:8089 TRINO_USER: ${TRINO_USER:-mo} + OPENMETADATA_URL: ${OPENMETADATA_URL:-http://10.0.21.47:8585} KAFKA_UI_URL: http://10.0.21.36:9000 HDFS_NN_URL: http://10.0.21.61:9870 S3_ENDPOINT: ${S3_ENDPOINT:-http://10.0.20.111:9020}