From 0402a4fbae6f1e888fd5ef4fe857ad9313b068bc Mon Sep 17 00:00:00 2001 From: Lakehouse Admin Date: Sun, 28 Jun 2026 17:08:57 +0200 Subject: [PATCH] update (bart) --- compose/forgejo/compose.yaml | 19 - compose/lam/compose.yml | 17 - config/docker/atc-db02/docker-compose.yml | 13 + config/docker/atc-db02/init/mongo/01-users.js | 15 + config/homepage/bookmarks.yaml | 4 + config/homepage/services.yaml | 35 +- .../homepage/services.yaml.bak-20260603234522 | 635 ++++++++++++++++++ config/homepage/widgets.yaml | 8 +- config/kafka/mongodb-connector.json | 11 +- config/scripts/collect_pipeline_metrics.py | 264 ++++++++ config/superset/Dockerfile | 8 +- .../superset/apply_palantir_rich_dashboard.py | 453 +++++++++++++ config/superset/bootstrap_lakehouse.py | 95 +++ .../superset/bootstrap_pipeline_dashboard.py | 367 ++++++++++ config/superset/collect_pipeline_metrics.py | 264 ++++++++ config/superset/fix_charts_qc.py | 85 +++ config/superset/palantir_dashboard.css | 97 +++ config/superset/save_query_context.py | 15 + config/superset/superset_config.py | 134 +++- config/trino/catalog/kafka.properties | 4 + .../catalog/mongodb_supplychain.properties | 2 +- 21 files changed, 2471 insertions(+), 74 deletions(-) delete mode 100644 compose/forgejo/compose.yaml delete mode 100644 compose/lam/compose.yml create mode 100644 config/docker/atc-db02/init/mongo/01-users.js create mode 100644 config/homepage/services.yaml.bak-20260603234522 create mode 100644 config/scripts/collect_pipeline_metrics.py create mode 100644 config/superset/apply_palantir_rich_dashboard.py create mode 100644 config/superset/bootstrap_lakehouse.py create mode 100644 config/superset/bootstrap_pipeline_dashboard.py create mode 100644 config/superset/collect_pipeline_metrics.py create mode 100644 config/superset/fix_charts_qc.py create mode 100644 config/superset/palantir_dashboard.css create mode 100644 config/superset/save_query_context.py create mode 100644 config/trino/catalog/kafka.properties diff --git a/compose/forgejo/compose.yaml b/compose/forgejo/compose.yaml deleted file mode 100644 index 58d1b38..0000000 --- a/compose/forgejo/compose.yaml +++ /dev/null @@ -1,19 +0,0 @@ -# Forgejo (lokaal op atc-docker01) — primaire Git staat op atc-mgt01:3001 -services: - forgejo: - image: codeberg.org/forgejo/forgejo:14 - container_name: forgejo - restart: always - environment: - USER_UID: "1000" - USER_GID: "1000" - ports: - - "4002:3000" - - "222:22" - volumes: - - forgejo_data:/data - - /etc/localtime:/etc/localtime:ro - -volumes: - forgejo_data: - name: forgejo_forgejo diff --git a/compose/lam/compose.yml b/compose/lam/compose.yml deleted file mode 100644 index dd02c75..0000000 --- a/compose/lam/compose.yml +++ /dev/null @@ -1,17 +0,0 @@ -# LDAP Account Manager (lokaal op atc-docker01) -# Productie LDAP/LAM: http://atc-mgt01.dell-atc.lan/lam/ -services: - lam: - image: ghcr.io/ldapaccountmanager/lam:stable - container_name: lam-app-1 - restart: unless-stopped - ports: - - "4001:80" - environment: - DEBIAN_FRONTEND: noninteractive - volumes: - - lam_data:/var/lib/ldap-account-manager - -volumes: - lam_data: - name: lam_lam diff --git a/config/docker/atc-db02/docker-compose.yml b/config/docker/atc-db02/docker-compose.yml index 022c45c..0a536a3 100644 --- a/config/docker/atc-db02/docker-compose.yml +++ b/config/docker/atc-db02/docker-compose.yml @@ -35,6 +35,19 @@ services: - "27017:27017" volumes: - mongodb_supplychain_data:/data/db + - ./init/mongo:/docker-entrypoint-initdb.d:ro + + mongo-express: + image: mongo-express + container_name: mongo_express + depends_on: + - mongodb-supplychain + ports: + - "8081:8081" + environment: + ME_CONFIG_MONGODB_URL: mongodb://mo:Dell2026%21@mongodb_supplychain:27017/?authSource=admin + ME_CONFIG_MONGODB_ENABLE_ADMIN: "true" + ME_CONFIG_BASICAUTH: "false" cassandra-telemetry: image: cassandra:4.1 diff --git a/config/docker/atc-db02/init/mongo/01-users.js b/config/docker/atc-db02/init/mongo/01-users.js new file mode 100644 index 0000000..a130509 --- /dev/null +++ b/config/docker/atc-db02/init/mongo/01-users.js @@ -0,0 +1,15 @@ +// Idempotent admin users for lab (mo + bart) +db = db.getSiblingDB("admin"); +var pw = "Dell2026!"; +["mo", "bart"].forEach(function(name) { + if (db.getUser(name) == null) { + db.createUser({ + user: name, + pwd: pw, + roles: [{ role: "root", db: "admin" }] + }); + print("Created user: " + name); + } else { + print("User exists: " + name); + } +}); diff --git a/config/homepage/bookmarks.yaml b/config/homepage/bookmarks.yaml index 199f96b..dcf6322 100644 --- a/config/homepage/bookmarks.yaml +++ b/config/homepage/bookmarks.yaml @@ -9,6 +9,10 @@ - Proxmox: - abbr: PVE href: https://10.0.10.65:8006/ + - Dockhand: + - abbr: DH + href: http://atc-docker01.dell-atc.lan:8082/ + - RSS Feeds: - TLDR Data Engineering: diff --git a/config/homepage/services.yaml b/config/homepage/services.yaml index ad8eec7..fc9db2a 100644 --- a/config/homepage/services.yaml +++ b/config/homepage/services.yaml @@ -124,6 +124,26 @@ description: atc-mgt01.dell-atc.lan:3001 siteMonitor: http://atc-mgt01.dell-atc.lan:3001/ color: "#F05032" + - GPU Lab: + icon: mdi-gpu + href: http://10.0.20.106:9000/ + description: atc-gpu-dev VM303 · 4× V100 · model manager + chat + siteMonitor: http://10.0.20.106:9000/ + color: "#76B900" + - Dockhand: + icon: mdi-docker + href: http://atc-docker01.dell-atc.lan:8082/ + description: Docker control plane · GPU-Dev + Bart-GPU + Mo-GPU + siteMonitor: http://atc-docker01.dell-atc.lan:8082/api/health + color: "#2496ED" + + - ATC Command Center: + icon: mdi-robot-outline + href: http://10.0.21.33/ + description: VM304 MCP · agent hub & ops floor · 10.0.21.33 + siteMonitor: http://10.0.21.33/ + color: "#0099cc" + - LDAP Admin: icon: http://atc-docker01.dell-atc.lan:8080/ldap.png href: http://atc-mgt01.dell-atc.lan/lam/ @@ -142,11 +162,22 @@ ping: atc-db01.dell-atc.lan description: atc-db01:3306 · mysql://USER@atc-db01.dell-atc.lan:3306/DB · mysql -h atc-db01 color: "#4479A1" + - Mongo Express · db02: + icon: http://atc-docker01.dell-atc.lan:8080/mongodb.svg + href: http://atc-db02.dell-atc.lan:8081/ + description: Web UI · supplychain · 3M events · users mo & bart + siteMonitor: http://atc-db02.dell-atc.lan:8081/ + ping: atc-db02.dell-atc.lan + color: "#47A248" - MongoDB · db02: icon: http://atc-docker01.dell-atc.lan:8080/mongodb.svg ping: atc-db02.dell-atc.lan - description: atc-db02:27017 · mongodb://USER@atc-db02.dell-atc.lan:27017/DB · mongosh - color: "#47A248" + description: | + :27017 · mongodb_supplychain container + DB supplychain · collection events (~3M) + mongosh: mongodb://mo@atc-db02.dell-atc.lan:27017/supplychain?authSource=admin + Trino: mongodb_supplychain.supplychain.events + color: "#3d8b40" - PostgreSQL · db02: icon: http://atc-docker01.dell-atc.lan:8080/postgresql.svg ping: atc-db02.dell-atc.lan diff --git a/config/homepage/services.yaml.bak-20260603234522 b/config/homepage/services.yaml.bak-20260603234522 new file mode 100644 index 0000000..0b0e465 --- /dev/null +++ b/config/homepage/services.yaml.bak-20260603234522 @@ -0,0 +1,635 @@ +--- +# ATC Lakehouse — Dell Technologies FDE Dashboard (Bart & Mo) + + +- Lakehouse · Architecture: + - Environment Map: + icon: mdi-sitemap + href: http://atc-docker01.dell-atc.lan:8080/docs/architecture.html + description: High-level diagram — data flow, hosts, and service map + color: "#007DB8" + - Git Docs: + icon: mdi-book-open-page-variant + href: http://atc-mgt01.dell-atc.lan:3001/mo/Lakehouse/src/branch/master/docs/landscape.md + description: Application landscape (Markdown in Forgejo) + color: "#E8752A" + +- Data Pipeline: + - Kafka UI: + icon: http://atc-docker01.dell-atc.lan:8080/apachekafka.svg + href: http://atc-kafka01.dell-atc.lan:9000/ + description: atc-kafka01.dell-atc.lan:9000 + siteMonitor: http://atc-kafka01.dell-atc.lan:9000/ + color: "#E02014" + - Debezium: + icon: http://atc-docker01.dell-atc.lan:8080/debezium.png + href: http://atc-lake01.dell-atc.lan:8083/ + description: atc-lake01.dell-atc.lan:8083 + siteMonitor: http://atc-lake01.dell-atc.lan:8083/ + color: "#4ECDC4" + - Spark: + icon: http://atc-docker01.dell-atc.lan:8080/apachespark.svg + href: http://atc-lake01.dell-atc.lan:8080/ + description: atc-lake01.dell-atc.lan:8080 + siteMonitor: http://atc-lake01.dell-atc.lan:8080/ + color: "#E25A1C" + - Trino: + icon: http://atc-docker01.dell-atc.lan:8080/trino.svg + href: http://atc-lake01.dell-atc.lan:8089/ui/ + description: atc-lake01.dell-atc.lan:8089/ui/ + siteMonitor: http://atc-lake01.dell-atc.lan:8089/ui/ + color: "#DD00A1" + - Airflow: + icon: http://atc-docker01.dell-atc.lan:8080/apacheairflow.svg + href: http://10.0.21.55:8080/ + description: 10.0.21.55:8080 + siteMonitor: http://10.0.21.55:8080/health + color: "#017CEE" + +- Analytics: + - Superset: + icon: http://atc-docker01.dell-atc.lan:8080/apachesuperset.svg + href: http://atc-docker01.dell-atc.lan:8088/ + description: atc-docker01.dell-atc.lan:8088 + siteMonitor: http://atc-docker01.dell-atc.lan:8088/health + color: "#6c5ce7" + - Kibana: + icon: http://atc-docker01.dell-atc.lan:8080/kibana.svg + href: http://atc-elastic01.dell-atc.lan:5601/ + description: atc-elastic01.dell-atc.lan:5601 + siteMonitor: http://atc-elastic01.dell-atc.lan:5601/ + color: "#F04E98" + - Elasticsearch: + icon: http://atc-docker01.dell-atc.lan:8080/elasticsearch.svg + ping: atc-elastic01.dell-atc.lan + description: | + Host: atc-elastic01.dell-atc.lan + Port: 9200 (no public HTTP) + curl: curl http://atc-elastic01.dell-atc.lan:9200 + Use Kibana for browser UI + color: "#005571" + - Grafana: + icon: http://atc-docker01.dell-atc.lan:8080/grafana.svg + href: http://atc-grafana.dell-atc.lan:3000/ + ping: atc-grafana.dell-atc.lan + description: | + Host: atc-grafana.dell-atc.lan + IP: 10.0.20.103 + Port: 3000 + Note: start with systemctl start grafana-server + siteMonitor: http://atc-grafana.dell-atc.lan:3000/ + color: "#F46800" + +- Infrastructure: + - ObjectScale UI: + icon: http://atc-docker01.dell-atc.lan:8080/dell.svg + href: https://10.0.20.111/ + description: ECS admin · https://10.0.20.111 · luna.local + siteMonitor: https://10.0.20.111/ + color: "#007DB8" + - ObjectScale S3: + icon: http://atc-docker01.dell-atc.lan:8080/minio.svg + href: http://10.0.20.111:9020/ + description: S3 API :9020 · bucket data · Trino Iceberg + Spark + siteMonitor: http://10.0.20.111:9020/ + color: "#C72C48" + - ObjectScale SSH: + icon: mdi-console + href: https://10.0.20.111/ + description: admin@atc-objectscale · appliance SSH · see docs/objectscale.md + ping: 10.0.20.111 + color: "#64748b" + - iDRAC: + icon: http://atc-docker01.dell-atc.lan:8080/dell.svg + href: https://10.0.41.102/ + description: Dell iDRAC · 10.0.41.102 + siteMonitor: https://10.0.41.102/ + color: "#007DB8" + - Proxmox: + icon: http://atc-docker01.dell-atc.lan:8080/proxmox.svg + href: https://10.0.10.65:8006/ + description: 10.0.10.65:8006 + siteMonitor: https://10.0.10.65:8006/ + color: "#E57000" + widget: + type: proxmox + url: https://10.0.10.65:8006 + username: "root@pam!homepage" + password: "8890185b-0850-42b7-bab2-a690ea4dc3f1" + node: pve01 + fields: ["vms", "lxc", "resources.cpu", "resources.mem"] + - Forgejo: + icon: http://atc-docker01.dell-atc.lan:8080/git.svg + href: http://atc-mgt01.dell-atc.lan:3001/ + description: atc-mgt01.dell-atc.lan:3001 + siteMonitor: http://atc-mgt01.dell-atc.lan:3001/ + color: "#F05032" + - LDAP Admin: + icon: http://atc-docker01.dell-atc.lan:8080/ldap.png + href: http://atc-mgt01.dell-atc.lan/lam/ + description: atc-mgt01.dell-atc.lan/lam/ + siteMonitor: http://atc-mgt01.dell-atc.lan/lam/ + color: "#0984e3" + +- Databases: + - PostgreSQL · db01: + icon: http://atc-docker01.dell-atc.lan:8080/postgresql.svg + ping: atc-db01.dell-atc.lan + description: atc-db01:5432 · postgresql://USER@atc-db01.dell-atc.lan:5432/DB · psql -h atc-db01 + color: "#4169E1" + - MySQL · db01: + icon: http://atc-docker01.dell-atc.lan:8080/mysql.svg + ping: atc-db01.dell-atc.lan + description: atc-db01:3306 · mysql://USER@atc-db01.dell-atc.lan:3306/DB · mysql -h atc-db01 + color: "#4479A1" + - Mongo Express · db02: + icon: http://atc-docker01.dell-atc.lan:8080/mongodb.svg + href: http://atc-db02.dell-atc.lan:8081/ + description: Web UI · supplychain · 3M events · users mo & bart + siteMonitor: http://atc-db02.dell-atc.lan:8081/ + ping: atc-db02.dell-atc.lan + color: "#47A248" + - MongoDB · db02: + icon: http://atc-docker01.dell-atc.lan:8080/mongodb.svg + ping: atc-db02.dell-atc.lan + description: | + :27017 · mongodb_supplychain container + DB supplychain · collection events (~3M) + mongosh: mongodb://mo@atc-db02.dell-atc.lan:27017/supplychain?authSource=admin + Trino: mongodb_supplychain.supplychain.events + color: "#3d8b40" + - PostgreSQL · db02: + icon: http://atc-docker01.dell-atc.lan:8080/postgresql.svg + ping: atc-db02.dell-atc.lan + description: atc-db02:5432 · postgresql://USER@atc-db02.dell-atc.lan:5432/DB · psql -h atc-db02 + color: "#336791" + - MySQL · db02: + icon: http://atc-docker01.dell-atc.lan:8080/mysql.svg + ping: atc-db02.dell-atc.lan + description: atc-db02:3306 · mysql://USER@atc-db02.dell-atc.lan:3306/DB · mysql -h atc-db02 + color: "#00758F" + - Cassandra · db02: + icon: http://atc-docker01.dell-atc.lan:8080/apachecassandra.svg + ping: atc-db02.dell-atc.lan + description: atc-db02:9042 · cqlsh atc-db02.dell-atc.lan 9042 · cassandra://atc-db02:9042 + color: "#1287B1" + - Neo4j · db02: + icon: http://atc-docker01.dell-atc.lan:8080/neo4j.svg + href: http://atc-db02.dell-atc.lan:7474/ + ping: atc-db02.dell-atc.lan + description: HTTP :7474 · bolt://atc-db02:7687 · neo4j://atc-db02.dell-atc.lan:7687 + siteMonitor: http://atc-db02.dell-atc.lan:7474/ + color: "#008CC1" + +- Intel · Hacker News: + - HN Front Page: + icon: mdi-newspaper + href: https://news.ycombinator.com/ + description: Hacker News — front page + color: "#E8752A" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/hn-front + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - HN Data Engineering: + icon: mdi-database-search + href: https://hn.algolia.com/?q=data%20engineering + description: Hacker News — data engineering + color: "#3b82f6" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/hn-de + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - HN Kafka: + icon: http://atc-docker01.dell-atc.lan:8080/apachekafka.svg + href: https://hn.algolia.com/?q=kafka + description: Hacker News — Kafka & streaming + color: "#E02014" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/hn-kafka + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - HN Spark: + icon: http://atc-docker01.dell-atc.lan:8080/apachespark.svg + href: https://hn.algolia.com/?q=apache+spark + description: Hacker News — Apache Spark + color: "#E25A1C" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/hn-spark + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Lobsters: + icon: mdi-lobster + href: https://lobste.rs/ + description: Computing & infra — curated links + color: "#ef4444" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/lobsters + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + +- Intel · Data Architecture: + - Data Eng Weekly: + icon: mdi-calendar-week + href: https://www.dataengineeringweekly.com/ + description: Weekly newsletter — pipelines & platforms + color: "#22d3ee" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/de-weekly + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Pragmatic Engineer: + icon: mdi-account-tie + href: https://blog.pragmaticengineer.com/ + description: Big tech engineering & architecture + color: "#a855f7" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/pragmatic + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Martin Fowler: + icon: mdi-arch + href: https://martinfowler.com/ + description: Software architecture & design + color: "#64748b" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/martinfowler + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - InfoQ: + icon: mdi-information-outline + href: https://www.infoq.com/ + description: Architecture, data & dev practices + color: "#94a3b8" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/infoq + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - ByteByteGo: + icon: mdi-school + href: https://blog.bytebytego.com/ + description: System design — scalable architectures + color: "#f59e0b" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/bytebytego + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - TLDR Data Eng: + icon: mdi-lightning-bolt + href: https://tldr.tech/dataengineering/ + description: Daily data engineering digest + color: "#E8752A" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/tldr-de + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + + + - Seattle Data Guy: + icon: mdi-chart-bar + href: https://www.seattledataguy.com/ + description: Practical data engineering tutorials + color: "#0ea5e9" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/seattle-de + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - RedMonk: + icon: mdi-chart-line + href: https://redmonk.com/ + description: Developer-focused industry analysis + color: "#dc2626" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/redmonk + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + +- Intel · Platforms & Blogs: + - Confluent Blog: + icon: http://atc-docker01.dell-atc.lan:8080/apachekafka.svg + href: https://www.confluent.io/blog/ + description: Kafka & event streaming + color: "#007DB8" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/confluent + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Databricks Blog: + icon: mdi-layers-triple + href: https://www.databricks.com/blog + description: Lakehouse & Spark platform + color: "#fb923c" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/databricks + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Debezium CDC: + icon: http://atc-docker01.dell-atc.lan:8080/debezium.png + href: https://debezium.io/blog/ + description: Change data capture patterns + color: "#4ECDC4" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/debezium + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - AWS Big Data: + icon: mdi-aws + href: https://aws.amazon.com/blogs/big-data/ + description: AWS analytics & data lakes + color: "#FF9900" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/aws-bigdata + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Google Cloud Blog: + icon: mdi-google-cloud + href: https://cloud.google.com/blog + description: GCP data & analytics + color: "#4285F4" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/google-cloud + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Cloudflare Eng: + icon: mdi-cloud + href: https://blog.cloudflare.com/ + description: Edge, networking & scale + color: "#F38020" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/cloudflare + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Meta Engineering: + icon: mdi-facebook + href: https://engineering.fb.com/ + description: Large-scale infra & data + color: "#1877F2" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/fb-engineering + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + + - Airflow Blog: + icon: http://atc-docker01.dell-atc.lan:8080/apacheairflow.svg + href: https://airflow.apache.org/blog/ + description: Workflow orchestration & DAGs + color: "#017CEE" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/airflow + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - Elastic Blog: + icon: http://atc-docker01.dell-atc.lan:8080/elasticsearch.svg + href: https://www.elastic.co/blog/ + description: Search, observability & security + color: "#005571" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/elastic + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + - The New Stack: + icon: mdi-newspaper-variant-multiple + href: https://thenewstack.io/ + description: Cloud native & platform engineering + color: "#22c55e" + widget: + type: customapi + url: http://atc-docker01.dell-atc.lan:8090/feed/thenewstack + refreshInterval: 300000 + display: dynamic-list + mappings: + items: items + name: title + href: link + target: _blank + + +- Documentation: + - Architecture Diagram: + icon: mdi-sitemap + href: http://atc-docker01.dell-atc.lan:8080/docs/architecture.html + description: High-level environment map (Mermaid) + color: "#007DB8" + - Application Landscape: + icon: mdi-book-open-variant + href: http://atc-mgt01.dell-atc.lan:3001/mo/Lakehouse/src/branch/master/docs/landscape.md + description: docs/landscape.md in Forgejo + color: "#E8752A" + - Network Diagram: + icon: mdi-lan + href: http://atc-mgt01.dell-atc.lan:3001/mo/Lakehouse/src/branch/master/docs/network.md + description: Subnets, ports, SSH mesh + color: "#22d3ee" + - Disaster Recovery: + icon: mdi-backup-restore + href: http://atc-mgt01.dell-atc.lan:3001/mo/Lakehouse/src/branch/master/docs/disaster-recovery.md + description: Restore procedures from git + color: "#a855f7" + - Docker Inventory: + icon: mdi-docker + href: http://atc-mgt01.dell-atc.lan:3001/mo/Lakehouse/src/branch/master/docs/docker-inventory.md + description: Containers per host + color: "#2496ED" + - Proxmox VMs: + icon: mdi-server + href: http://atc-mgt01.dell-atc.lan:3001/mo/Lakehouse/src/branch/master/inventory/proxmox-vms.json + description: VM inventory (JSON) + color: "#E57000" + +- News & Resources: + - Dell Technologies: + icon: http://atc-docker01.dell-atc.lan:8080/dell-technologies.svg + href: https://www.delltechnologies.com/ + description: Corporate site & solutions + color: "#007DB8" + - Dell Blog · Data: + icon: mdi-post-outline + href: https://www.dell.com/en-us/blog/categories/products-solutions-analytics + description: Analytics & data solutions + color: "#007DB8" + - Apache Spark: + icon: http://atc-docker01.dell-atc.lan:8080/apachespark.svg + href: https://spark.apache.org/docs/latest/ + description: Spark documentation + color: "#E25A1C" + - Trino: + icon: http://atc-docker01.dell-atc.lan:8080/trino.svg + href: https://trino.io/docs/current/ + description: Distributed SQL engine docs + color: "#DD00A1" + - Apache Iceberg: + icon: mdi-snowflake + href: https://iceberg.apache.org/docs/latest/ + description: Open table format for lakes + color: "#38bdf8" + - Real Python: + icon: mdi-language-python + href: https://realpython.com/ + description: Python tutorials & patterns + color: "#3776AB" + - MinIO Docs: + icon: http://atc-docker01.dell-atc.lan:8080/minio.svg + href: https://min.io/docs/minio/linux/index.html + description: S3-compatible object storage + color: "#C72C48" + - TLDR Tech: + icon: mdi-lightning-bolt-outline + href: https://tldr.tech/ + description: Daily tech newsletter (web) + color: "#E8752A" + +- Docker · atc-docker01: + - Homepage: + icon: mdi-view-dashboard + href: http://atc-docker01.dell-atc.lan/ + description: atc-docker01.dell-atc.lan + server: my-docker + container: homepage + color: "#E8752A" + - Superset: + icon: http://atc-docker01.dell-atc.lan:8080/apachesuperset.svg + href: http://atc-docker01.dell-atc.lan:8088/ + description: atc-docker01.dell-atc.lan:8088 + server: my-docker + container: superset + color: "#6c5ce7" + - Forgejo Local: + icon: http://atc-docker01.dell-atc.lan:8080/git.svg + href: http://atc-docker01.dell-atc.lan:4002/ + description: atc-docker01.dell-atc.lan:4002 + siteMonitor: http://atc-docker01.dell-atc.lan:4002/ + server: my-docker + container: forgejo + color: "#F05032" diff --git a/config/homepage/widgets.yaml b/config/homepage/widgets.yaml index a99fec4..7054199 100644 --- a/config/homepage/widgets.yaml +++ b/config/homepage/widgets.yaml @@ -25,9 +25,9 @@ units: metric refresh: 3000 - openmeteo: - label: Lab - latitude: 52.37 - longitude: 4.89 + label: Amsterdam + latitude: 52.374 + longitude: 4.890 timezone: Europe/Amsterdam units: metric - cache: 30 + cache: 15 diff --git a/config/kafka/mongodb-connector.json b/config/kafka/mongodb-connector.json index 5bba347..1335373 100644 --- a/config/kafka/mongodb-connector.json +++ b/config/kafka/mongodb-connector.json @@ -1 +1,10 @@ -{"connector.class":"io.debezium.connector.mongodb.MongoDbConnector","topic.prefix":"mongodb-supplychain","mongodb.history.kafka.bootstrap.servers":"localhost:9092","mongodb.history.kafka.topic":"schema-changes.supplychain","mongodb.connection.string":"mongodb://10.0.21.51:27017","name":"mongodb-connector","mongodb.name":"supplychain","snapshot.mode":"initial"} \ No newline at end of file +{ + "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", + "topic.prefix": "mongodb-supplychain", + "mongodb.connection.string": "mongodb://mo:Dell2026%21@10.0.21.51:27017/?authSource=admin", + "mongodb.history.kafka.bootstrap.servers": "10.0.21.36:9092", + "mongodb.history.kafka.topic": "schema-changes.supplychain", + "name": "mongodb-connector", + "mongodb.name": "supplychain", + "snapshot.mode": "initial" +} diff --git a/config/scripts/collect_pipeline_metrics.py b/config/scripts/collect_pipeline_metrics.py new file mode 100644 index 0000000..4f81df6 --- /dev/null +++ b/config/scripts/collect_pipeline_metrics.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Collect Debezium, Kafka CDC, and Spark metrics into postgres monitor schema.""" +import json +import shlex +import subprocess +import urllib.request +from collections import Counter +from datetime import datetime, timezone + +import psycopg2 + +PG_DSN = "host=10.0.21.51 dbname=postgres user=mo password=Dell2026!" +KAFKA = "10.0.21.36:9092" +DEBEZIUM = "http://localhost:8083" # Kafka Connect on kafka01; fallback lake01 :8083 +SPARK_MASTER = "http://10.0.21.50:8080" + +CDC_TOPICS = [ + ("PostgreSQL", "postgres-sales.public.sales_orders"), + ("MongoDB", "mongodb-supplychain.supplychain.events"), +] + +OP_LABELS = {"c": "INSERT", "u": "UPDATE", "d": "DELETE", "r": "SNAPSHOT", "i": "INSERT"} + + +def fetch_json(url, timeout=10): + with urllib.request.urlopen(url, timeout=timeout) as r: + return json.loads(r.read().decode()) + + +def collect_debezium(cur): + connectors = fetch_json(f"{DEBEZIUM}/connectors") + cur.execute("DELETE FROM monitor.debezium_connectors") + now = datetime.now(timezone.utc) + for name in connectors: + try: + st = fetch_json(f"{DEBEZIUM}/connectors/{name}/status") + except Exception as e: + cur.execute( + """INSERT INTO monitor.debezium_connectors + (connector_name, state, task_state, worker_id, checked_at) + VALUES (%s,%s,%s,%s,%s)""", + (name, "ERROR", str(e)[:32], "", now), + ) + continue + conn_state = st.get("connector", {}).get("state", "UNKNOWN") + tasks = st.get("tasks") or [] + task_state = tasks[0].get("state", "NONE") if tasks else "NONE" + worker = st.get("connector", {}).get("worker_id", "") + cur.execute( + """INSERT INTO monitor.debezium_connectors + (connector_name, state, task_state, worker_id, checked_at) + VALUES (%s,%s,%s,%s,%s)""", + (name, conn_state, task_state, worker, now), + ) + + +KAFKA_BIN = "/opt/kafka/bin" +USE_SSH_KAFKA = False # set True when running off-host + + +def _kafka_cmd(bin_name, args): + parts = [f"{KAFKA_BIN}/{bin_name}"] + list(args) + if USE_SSH_KAFKA: + remote = " ".join(shlex.quote(p) for p in parts) + full = f"ssh -o StrictHostKeyChecking=no root@10.0.21.36 {remote}" + return subprocess.check_output(full, shell=True, stderr=subprocess.DEVNULL, timeout=90, text=True) + return subprocess.check_output(parts, stderr=subprocess.DEVNULL, timeout=90, text=True) + + +def kafka_end_offsets(topic): + try: + out = _kafka_cmd( + "kafka-run-class.sh", + [ + "kafka.tools.GetOffsetShell", + "--broker-list", + "localhost:9092", + "--topic", + topic, + ], + ) + except Exception: + return [] + rows = [] + for line in out.strip().splitlines(): + parts = line.split(":") + if len(parts) >= 3: + rows.append((int(parts[1]), int(parts[2]))) + return rows + + +def sample_topic_messages(topic, max_msgs=3000, tail=5000): + """Sample recent messages using kafka-console-consumer from tail.""" + offsets = kafka_end_offsets(topic) + if not offsets: + return [] + # Pick partition 0 for sampling + part, end = offsets[0] + start = max(0, end - tail) + try: + out = _kafka_cmd( + "kafka-console-consumer.sh", + [ + "--bootstrap-server", + "localhost:9092", + "--topic", + topic, + "--partition", + str(part), + "--offset", + str(start), + "--max-messages", + str(min(max_msgs, tail)), + "--timeout-ms", + "15000", + ], + ) + except Exception: + return [] + return [ln for ln in out.strip().split("\n") if ln.strip()] + + +def parse_debezium_line(line): + try: + doc = json.loads(line) + payload = doc.get("payload") or doc + op = payload.get("op") or payload.get("operationType") or "?" + src = payload.get("source") or {} + table = src.get("table") or src.get("collection") or "" + ts_ms = payload.get("ts_ms") or src.get("ts_ms") + after = payload.get("after") or {} + before = payload.get("before") or {} + row = after if after else before + key = str(row.get("order_id") or row.get("event_id") or row.get("_id") or "")[:200] + detail = str(row.get("region") or row.get("type") or row.get("department") or "")[:200] + event_ts = None + if ts_ms: + event_ts = datetime.fromtimestamp(int(ts_ms) / 1000, tz=timezone.utc) + return op, table, key, detail, event_ts + except Exception: + return None + + +def collect_kafka_cdc(cur): + now = datetime.now(timezone.utc) + cur.execute("DELETE FROM monitor.kafka_topics") + cur.execute("DELETE FROM monitor.cdc_operations") + cur.execute("DELETE FROM monitor.cdc_recent_events") + + for source, topic in CDC_TOPICS: + for part, end in kafka_end_offsets(topic): + cur.execute( + """INSERT INTO monitor.kafka_topics (topic, partition_id, end_offset, checked_at) + VALUES (%s,%s,%s,%s)""", + (topic, part, end, now), + ) + + lines = sample_topic_messages(topic, max_msgs=2000, tail=3000) + ops = Counter() + recent = [] + for line in lines: + parsed = parse_debezium_line(line) + if not parsed: + continue + op, table, key, detail, event_ts = parsed + ops[op] += 1 + if len(recent) < 100: + recent.append((op, table, key, detail, event_ts)) + + for op, cnt in ops.items(): + cur.execute( + """INSERT INTO monitor.cdc_operations + (source_system, topic, operation, operation_label, event_count, checked_at) + VALUES (%s,%s,%s,%s,%s,%s)""", + (source, topic, op, OP_LABELS.get(op, op), cnt, now), + ) + + for op, table, key, detail, event_ts in recent[:50]: + cur.execute( + """INSERT INTO monitor.cdc_recent_events + (source_system, topic, operation, operation_label, table_name, + record_key, detail, event_ts, sampled_at) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)""", + ( + source, + topic, + op, + OP_LABELS.get(op, op), + table, + key, + detail, + event_ts, + now, + ), + ) + + +def collect_spark(cur): + now = datetime.now(timezone.utc) + cur.execute("DELETE FROM monitor.spark_applications") + try: + data = fetch_json(f"{SPARK_MASTER}/json/", timeout=5) + apps = [] + if isinstance(data, dict): + # Standalone master JSON + for a in data.get("activeapps", []) or []: + apps.append(a) + for a in data.get("completedapps", []) or []: + apps.append(a) + for a in apps[:20]: + cur.execute( + """INSERT INTO monitor.spark_applications + (app_id, app_name, state, cores, memory_mb, duration_sec, checked_at) + VALUES (%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (app_id) DO UPDATE SET + app_name=EXCLUDED.app_name, state=EXCLUDED.state, + cores=EXCLUDED.cores, memory_mb=EXCLUDED.memory_mb, + duration_sec=EXCLUDED.duration_sec, checked_at=EXCLUDED.checked_at""", + ( + a.get("id", "unknown"), + a.get("name", "Spark App"), + "RUNNING" if "attempts" not in a else "COMPLETED", + int(a.get("cores", 0) or 0), + int((a.get("memory", 0) or 0) / 1024 / 1024), + int(a.get("duration", 0) / 1000) if a.get("duration") else 0, + now, + ), + ) + except Exception as e: + # Placeholder row so dashboard shows Spark host status + cur.execute( + """INSERT INTO monitor.spark_applications + (app_id, app_name, state, cores, memory_mb, duration_sec, checked_at) + VALUES (%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (app_id) DO UPDATE SET state=EXCLUDED.state, checked_at=EXCLUDED.checked_at""", + ( + "spark-master", + f"Spark Master @ {SPARK_MASTER}", + "REACHABLE" if "Connection" not in str(e) else "UNREACHABLE", + 0, + 0, + 0, + now, + ), + ) + + +def main(): + conn = psycopg2.connect(PG_DSN) + conn.autocommit = True + cur = conn.cursor() + print("Collecting Debezium...") + collect_debezium(cur) + print("Collecting Kafka CDC samples...") + collect_kafka_cdc(cur) + print("Collecting Spark...") + collect_spark(cur) + cur.close() + conn.close() + print("Done.") + + +if __name__ == "__main__": + main() diff --git a/config/superset/Dockerfile b/config/superset/Dockerfile index 521c1a4..7ffce6c 100644 --- a/config/superset/Dockerfile +++ b/config/superset/Dockerfile @@ -1,3 +1,7 @@ FROM apache/superset:latest - -RUN pip install psycopg2-binary +USER root +RUN pip3 install --no-cache-dir --target=/app/.venv/lib/python3.10/site-packages --no-deps \ + sqlalchemy-trino==0.5.0 trino==0.337.0 && \ + pip3 install --no-cache-dir --target=/app/.venv/lib/python3.10/site-packages \ + requests lz4 orjson python-dateutil pytz tzlocal zstandard charset-normalizer idna urllib3 certifi six greenlet +USER superset diff --git a/config/superset/apply_palantir_rich_dashboard.py b/config/superset/apply_palantir_rich_dashboard.py new file mode 100644 index 0000000..23da881 --- /dev/null +++ b/config/superset/apply_palantir_rich_dashboard.py @@ -0,0 +1,453 @@ +#!/usr/bin/env python3 +"""Apply Palantir styling and enrich ATC Lakehouse Superset dashboard.""" +import json +import os +import requests + +BASE = "http://127.0.0.1:8088" +DASH_ID = 1 +CSS_PATH = "/tmp/palantir_dashboard.css" + +NEW_CHARTS = [ + ( + "Trino · PostgreSQL Sales", + "public", + "sales_orders", + "PostgreSQL · Orders by Channel", + "pie", + { + "metric": {"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "COUNT(*)"}, + "groupby": ["sales_channel"], + "row_limit": 10, + }, + ), + ( + "Trino · PostgreSQL Sales", + "public", + "sales_orders", + "PostgreSQL · Avg Order by Region", + "echarts_timeseries_bar", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "AVG(amount)", "label": "Avg Amount"} + ], + "groupby": ["region"], + "row_limit": 15, + }, + ), + ( + "Trino · PostgreSQL Sales", + "public", + "sales_orders", + "PostgreSQL · Status Breakdown", + "pie", + { + "metric": {"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "COUNT(*)"}, + "groupby": ["order_status"], + "row_limit": 10, + }, + ), + ( + "Trino · MySQL HR", + "hr", + "employee_events", + "MySQL HR · By Event Type", + "pie", + { + "metric": {"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "COUNT(*)"}, + "groupby": ["event_type"], + "row_limit": 12, + }, + ), + ( + "Trino · MySQL HR", + "hr", + "employee_events", + "MySQL HR · Events per Month", + "echarts_timeseries_line", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "Events"} + ], + "groupby": [ + { + "expressionType": "SQL", + "sqlExpression": "date_trunc('month', event_ts)", + "label": "Month", + } + ], + "row_limit": 24, + }, + ), + ( + "Trino · MongoDB Supply Chain", + "supplychain", + "events", + "MongoDB · Amount by Source", + "echarts_timeseries_bar", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "SUM(amount)", "label": "Total Amount"} + ], + "groupby": ["source"], + "row_limit": 10, + }, + ), + ( + "Trino · MongoDB Supply Chain", + "supplychain", + "events", + "MongoDB · Events per Month", + "echarts_timeseries_line", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "Events"} + ], + "groupby": [ + { + "expressionType": "SQL", + "sqlExpression": "date_trunc('month', ts)", + "label": "Month", + } + ], + "row_limit": 24, + }, + ), + ( + "Trino · Cassandra Telemetry", + "telemetry", + "device_metrics", + "Cassandra · Avg Metric Over Time", + "echarts_timeseries_line", + { + "metrics": [ + { + "expressionType": "SQL", + "sqlExpression": "AVG(metric_value)", + "label": "Avg Value", + } + ], + "groupby": [ + { + "expressionType": "SQL", + "sqlExpression": "date_trunc('day', metric_ts)", + "label": "Day", + } + ], + "row_limit": 30, + }, + ), +] + +SECTIONS = [ + ("HEADER", "ATC Lakehouse · Federated Data Platform", "Trino · PostgreSQL · MySQL · MongoDB · Cassandra · Dell Technologies FDE"), + ("PostgreSQL Sales", "30M orders · postgres_sales.public.sales_orders"), + ("MySQL HR", "569K events · mysql_hr.hr.employee_events"), + ("MongoDB Supply Chain", "3M events · mongodb_supplychain.supplychain.events"), + ("Cassandra Telemetry", "Device metrics · cassandra_telemetry.telemetry.device_metrics"), +] + + +def session(): + s = requests.Session() + r = s.post( + f"{BASE}/api/v1/security/login", + json={"username": "admin", "password": "admin", "provider": "db", "refresh": True}, + ) + r.raise_for_status() + h = { + "Authorization": "Bearer " + r.json()["access_token"], + "Content-Type": "application/json", + } + h["X-CSRFToken"] = s.get(f"{BASE}/api/v1/security/csrf_token/", headers=h).json()["result"] + h["Referer"] = BASE + return s, h + + +def get_db_map(s, h): + r = s.get(f"{BASE}/api/v1/database/", headers=h) + r.raise_for_status() + return {d["database_name"]: d["id"] for d in r.json().get("result", [])} + + +def get_or_create_dataset(s, h, db_id, schema, table): + r = s.get(f"{BASE}/api/v1/dataset/", headers=h) + for d in r.json().get("result", []): + if ( + d.get("table_name") == table + and d.get("schema") == schema + and d.get("database", {}).get("id") == db_id + ): + return d["id"] + r = s.post( + f"{BASE}/api/v1/dataset/", + headers=h, + json={"database": db_id, "schema": schema, "table_name": table}, + ) + r.raise_for_status() + return r.json()["id"] + + +def create_chart(s, h, name, ds_id, viz_type, params): + r = s.get(f"{BASE}/api/v1/chart/", headers=h) + for c in r.json().get("result", []): + if c.get("slice_name") == name: + return c["id"] + full = {"datasource": f"{ds_id}__table", "viz_type": viz_type, "row_limit": 1000, **params} + r = s.post( + f"{BASE}/api/v1/chart/", + headers=h, + json={ + "slice_name": name, + "viz_type": viz_type, + "datasource_id": ds_id, + "datasource_type": "table", + "params": json.dumps(full), + "owners": [1, 2, 3], + }, + ) + if r.status_code not in (200, 201): + raise RuntimeError(f"chart {name}: {r.text[:300]}") + return r.json()["id"] + + +def build_layout(chart_items): + """chart_items: list of (chart_id, name, section) or ('md', title, subtitle).""" + layout = { + "DASHBOARD_VERSION": "v2", + "ROOT_ID": {"type": "ROOT", "id": "ROOT_ID", "children": ["GRID_ID"]}, + "GRID_ID": {"type": "GRID", "id": "GRID_ID", "children": [], "parents": ["ROOT_ID"]}, + } + row_idx = 0 + + def add_row(): + nonlocal row_idx + row_idx += 1 + rid = f"ROW-{row_idx}" + layout["GRID_ID"]["children"].append(rid) + layout[rid] = { + "type": "ROW", + "id": rid, + "children": [], + "parents": ["ROOT_ID", "GRID_ID"], + "meta": {"background": "BACKGROUND_TRANSPARENT"}, + } + return rid + + for item in chart_items: + if item[0] == "md": + _, title, subtitle = item + rid = add_row() + mid = f"MARKDOWN-{row_idx}" + layout[rid]["children"].append(mid) + layout[mid] = { + "type": "MARKDOWN", + "id": mid, + "children": [], + "parents": ["ROOT_ID", "GRID_ID", rid], + "meta": { + "width": 12, + "height": 12, + "code": f"## {title}\n\n{subtitle}", + }, + } + else: + cid, name, _section = item + rid = add_row() + # up to 3 charts per row + existing = [ + k + for k in layout[rid]["children"] + if k.startswith("CHART-") + ] + if len(existing) >= 3: + rid = add_row() + chart_key = f"CHART-explore-{cid}" + layout[rid]["children"].append(chart_key) + col = len([k for k in layout[rid]["children"] if k.startswith("CHART-")]) - 1 + layout[chart_key] = { + "type": "CHART", + "id": chart_key, + "children": [], + "parents": ["ROOT_ID", "GRID_ID", rid], + "meta": { + "width": 4, + "height": 55 if "Total" in name or "Records" in name else 65, + "chartId": cid, + "sliceName": name, + }, + } + + return layout + + +def save_query_contexts(): + app = __import__("superset.app", fromlist=["create_app"]).create_app() + with app.app_context(): + from flask import g + from superset.extensions import db + from superset.models.slice import Slice + from superset.charts.schemas import ChartDataQueryContextSchema + from superset import security_manager + + g.user = security_manager.find_user(username="admin") + for sl in db.session.query(Slice).all(): + try: + fd = sl.form_data + metric = fd.get("metric") + metrics = fd.get("metrics") or ([metric] if metric else []) + if not metrics: + metrics = [ + { + "expressionType": "SQL", + "sqlExpression": "COUNT(*)", + "label": "COUNT(*)", + } + ] + groupby = fd.get("groupby") or [] + payload = { + "datasource": {"id": sl.datasource_id, "type": sl.datasource_type}, + "force": False, + "queries": [ + { + "filters": [], + "extras": {"having": "", "where": ""}, + "applied_time_extras": {}, + "columns": groupby if isinstance(groupby, list) else [], + "metrics": metrics, + "orderby": [], + "annotation_layers": [], + "row_limit": int(fd.get("row_limit") or 1000), + "series_limit": 0, + "order_desc": True, + "url_params": {}, + "custom_params": {}, + "custom_form_data": {}, + } + ], + "form_data": fd, + "result_format": "json", + "result_type": "full", + } + ChartDataQueryContextSchema().load(payload) + sl.query_context = json.dumps(payload) + sl.query_context_generation = True + db.session.add(sl) + except Exception as e: + print("qc err", sl.id, e) + db.session.commit() + + +def main(): + s, h = session() + db_map = get_db_map(s, h) + + # Create new charts + new_ids = [] + for db_name, schema, table, name, viz, params in NEW_CHARTS: + db_id = db_map.get(db_name) + if not db_id: + print("skip, no db:", db_name) + continue + ds_id = get_or_create_dataset(s, h, db_id, schema, table) + cid = create_chart(s, h, name, ds_id, viz, params) + new_ids.append((cid, name, db_name.split("·")[-1].strip())) + print("new chart", cid, name) + + # All charts for dashboard + r = s.get(f"{BASE}/api/v1/chart/?q=(page:0,page_size:200)", headers=h) + all_charts = r.json().get("result", []) + + def sort_key(c): + n = c.get("slice_name") or "" + if "Lakehouse" in n or "Records per Source" in n: + return (0, n) + if "PostgreSQL" in n: + return (1, n) + if "MySQL" in n: + return (2, n) + if "MongoDB" in n: + return (3, n) + if "Cassandra" in n: + return (4, n) + return (5, n) + + all_charts.sort(key=sort_key) + + chart_items = [ + ("md", "ATC Lakehouse · Federated Data Platform", "Real-time analytics across all Trino catalogs · Dell Technologies"), + ] + current_section = None + for c in all_charts: + name = c.get("slice_name") or "" + if "PostgreSQL" in name and current_section != "pg": + chart_items.append(("md", "PostgreSQL Sales", "30M orders · CDC-enabled · atc-db02")) + current_section = "pg" + elif "MySQL" in name and current_section != "mysql": + chart_items.append(("md", "MySQL HR", "569K employee events · HR domain")) + current_section = "mysql" + elif "MongoDB" in name and current_section != "mongo": + chart_items.append(("md", "MongoDB Supply Chain", "3M supply chain events")) + current_section = "mongo" + elif "Cassandra" in name and current_section != "cass": + chart_items.append(("md", "Cassandra Telemetry", "IoT device metrics")) + current_section = "cass" + chart_items.append((c["id"], name, current_section)) + + chart_ids = [x[0] for x in chart_items if x[0] != "md"] + position = build_layout(chart_items) + + css = "" + if os.path.exists(CSS_PATH): + css = open(CSS_PATH, encoding="utf-8").read() + + chart_configuration = { + str(cid): {"id": cid, "crossFilters": {"scope": "global", "chartsInScope": chart_ids}} + for cid in chart_ids + } + + payload = { + "dashboard_title": "ATC Lakehouse · Trino Federated", + "published": True, + "position_json": json.dumps(position), + "css": css, + "json_metadata": json.dumps( + { + "color_scheme": "palantir_ops", + "label_colors": {}, + "refresh_frequency": 120, + "timed_refresh_immune_slices": [], + "expanded_slices": {}, + "chart_configuration": chart_configuration, + "global_chart_configuration": { + "scope": {"rootPath": ["ROOT_ID"], "excluded": []}, + "chartsInScope": chart_ids, + }, + "native_filter_configuration": [], + "color_scheme_domain": [], + "shared_label_colors": {}, + } + ), + "owners": [1, 2, 3], + } + + r = s.put(f"{BASE}/api/v1/dashboard/{DASH_ID}", headers=h, json=payload) + print("dashboard update", r.status_code) + if r.status_code >= 400: + print(r.text[:500]) + return + + for cid in chart_ids: + s.put( + f"{BASE}/api/v1/chart/{cid}", + headers=h, + json={"dashboards": [DASH_ID], "owners": [1, 2, 3]}, + ) + + print("Saving query contexts...") + save_query_contexts() + print(f"Done — {len(chart_ids)} charts, Palantir theme applied to dashboard.") + + +if __name__ == "__main__": + main() diff --git a/config/superset/bootstrap_lakehouse.py b/config/superset/bootstrap_lakehouse.py new file mode 100644 index 0000000..c44e2a8 --- /dev/null +++ b/config/superset/bootstrap_lakehouse.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +import json, sys, requests +BASE = "http://127.0.0.1:8088" +HOST = "10.0.21.50:8089" +USER = "mo" +SOURCES = [ + ("Trino · PostgreSQL Sales", f"trino://{USER}@{HOST}/postgres_sales/public", "public", "sales_orders", [ + ("PostgreSQL · Total Orders", "big_number_total", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"}}), + ("PostgreSQL · Revenue by Region", "pie", {"metric":{"expressionType":"SQL","sqlExpression":"SUM(amount)","label":"Revenue"},"groupby":["region"],"row_limit":20}), + ("PostgreSQL · Orders per Month", "echarts_timeseries_bar", {"metrics":[{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"Orders"}],"groupby":[{"expressionType":"SQL","sqlExpression":"date_trunc('month', order_ts)","label":"Month"}],"row_limit":24}), + ]), + ("Trino · MySQL HR", f"trino://{USER}@{HOST}/mysql_hr/hr", "hr", "employee_events", [ + ("MySQL HR · Total Events", "big_number_total", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"}}), + ("MySQL HR · By Department", "pie", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"},"groupby":["department"],"row_limit":15}), + ("MySQL HR · By Region", "echarts_timeseries_bar", {"metrics":[{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"Events"}],"groupby":["region"],"row_limit":20}), + ]), + ("Trino · MongoDB Supply Chain", f"trino://{USER}@{HOST}/mongodb_supplychain/supplychain", "supplychain", "events", [ + ("MongoDB · Total Events", "big_number_total", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"}}), + ("MongoDB · By Type", "pie", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"},"groupby":["type"],"row_limit":10}), + ("MongoDB · By Region", "echarts_timeseries_bar", {"metrics":[{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"Events"}],"groupby":["region"],"row_limit":10}), + ]), + ("Trino · Cassandra Telemetry", f"trino://{USER}@{HOST}/cassandra_telemetry/telemetry", "telemetry", "device_metrics", [ + ("Cassandra · Total Metrics", "big_number_total", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"}}), + ("Cassandra · By Metric Type", "pie", {"metric":{"expressionType":"SQL","sqlExpression":"COUNT(*)","label":"COUNT(*)"},"groupby":["metric_type"],"row_limit":20}), + ("Cassandra · Avg by Device", "echarts_timeseries_bar", {"metrics":[{"expressionType":"SQL","sqlExpression":"AVG(metric_value)","label":"Avg"}],"groupby":["device_id"],"row_limit":20}), + ]), +] +OVERVIEW_SQL = """SELECT 'PostgreSQL sales_orders' AS source, COUNT(*) AS records FROM postgres_sales.public.sales_orders +UNION ALL SELECT 'MySQL employee_events', COUNT(*) FROM mysql_hr.hr.employee_events +UNION ALL SELECT 'MongoDB events', COUNT(*) FROM mongodb_supplychain.supplychain.events +UNION ALL SELECT 'Cassandra device_metrics', COUNT(*) FROM cassandra_telemetry.telemetry.device_metrics""" + +def headers(s): + r=s.post(f"{BASE}/api/v1/security/login",json={"username":"admin","password":"admin","provider":"db","refresh":True}); r.raise_for_status() + h={"Authorization":"Bearer "+r.json()["access_token"],"Content-Type":"application/json"} + h["X-CSRFToken"]=s.get(f"{BASE}/api/v1/security/csrf_token/",headers=h).json()["result"]; h["Referer"]=BASE; return h + +def get_db(s,h,name,uri): + r=s.get(f"{BASE}/api/v1/database/",headers=h); r.raise_for_status() + for d in r.json().get("result",[]): + if d["database_name"]==name: return d["id"] + r=s.post(f"{BASE}/api/v1/database/",headers=h,json={"database_name":name,"sqlalchemy_uri":uri,"expose_in_sqllab":True,"allow_run_async":True}) + if r.status_code not in (200,201): raise SystemExit(r.text) + print("DB",name,r.json()["id"]); return r.json()["id"] + +def get_ds(s,h,db,schema,table,sql=None): + r=s.get(f"{BASE}/api/v1/dataset/",headers=h); r.raise_for_status() + for d in r.json().get("result",[]): + if d.get("table_name")==table and d.get("schema")==schema and d.get("database",{}).get("id")==db: return d["id"] + p={"database":db,"table_name":table,"schema":schema} if not sql else {"database":db,"table_name":table,"sql":sql} + r=s.post(f"{BASE}/api/v1/dataset/",headers=h,json=p) + if r.status_code not in (200,201): raise SystemExit(r.text) + print(" DS",table,r.json()["id"]); return r.json()["id"] + +def mk_chart(s,h,name,ds,viz,params): + r=s.get(f"{BASE}/api/v1/chart/",headers=h); r.raise_for_status() + for c in r.json().get("result",[]): + if c.get("slice_name")==name: return c["id"] + p={"datasource":f"{ds}__table","viz_type":viz,"row_limit":1000,**params} + r=s.post(f"{BASE}/api/v1/chart/",headers=h,json={"slice_name":name,"viz_type":viz,"datasource_id":ds,"datasource_type":"table","params":json.dumps(p)}) + if r.status_code not in (200,201): raise SystemExit(f"chart {name}: {r.text[:300]}") + print(" chart",name,r.json()["id"]); return r.json()["id"] + +def mk_dash(s,h,title,cids): + layout={"DASHBOARD_VERSION":"v2","ROOT_ID":{"type":"ROOT","id":"ROOT_ID","children":["GRID_ID"]},"GRID_ID":{"type":"GRID","id":"GRID_ID","children":[],"parents":["ROOT_ID"]}} + row=col=0 + for cid in cids: + k=f"CHART-{cid}"; x=(col%3)*4; y=row*12 + layout[k]={"type":"CHART","id":k,"children":[],"meta":{"width":4,"height":10,"chartId":cid},"parents":["ROOT_ID","GRID_ID"]} + layout["GRID_ID"]["children"].append(k); col+=1 + if col%3==0: row+=1 + payload={"dashboard_title":title,"published":True,"position_json":json.dumps(layout),"json_metadata":"{}"} + r=s.get(f"{BASE}/api/v1/dashboard/",headers=h); r.raise_for_status() + for d in r.json().get("result",[]): + if d.get("dashboard_title")==title: + did=d["id"]; s.put(f"{BASE}/api/v1/dashboard/{did}",headers=h,json=payload) + for cid in cids: s.put(f"{BASE}/api/v1/chart/{cid}",headers=h,json={"dashboards":[did]}) + print("Dashboard",did); return did + r=s.post(f"{BASE}/api/v1/dashboard/",headers=h,json=payload) + if r.status_code not in (200,201): raise SystemExit(r.text) + did=r.json()["id"] + for cid in cids: s.put(f"{BASE}/api/v1/chart/{cid}",headers=h,json={"dashboards":[did]}) + print("Dashboard",did); return did + +def main(): + s=requests.Session(); h=headers(s) + cids=[] + odb=get_db(s,h,"Trino · Lakehouse Overview",f"trino://{USER}@{HOST}/postgres_sales/public") + ods=get_ds(s,h,odb,None,"lakehouse_counts",OVERVIEW_SQL) + cids.append(mk_chart(s,h,"Lakehouse · Records per Source",ods,"pie",{"metric":{"expressionType":"SQL","sqlExpression":"SUM(records)","label":"Records"},"groupby":["source"],"row_limit":10})) + for dbn,uri,sch,tbl,charts in SOURCES: + db=get_db(s,h,dbn,uri); ds=get_ds(s,h,db,sch,tbl) + for nm,viz,pr in charts: cids.append(mk_chart(s,h,nm,ds,viz,pr)) + mk_dash(s,h,"ATC Lakehouse · Trino Federated",cids) +if __name__=="__main__": main() diff --git a/config/superset/bootstrap_pipeline_dashboard.py b/config/superset/bootstrap_pipeline_dashboard.py new file mode 100644 index 0000000..a73dab9 --- /dev/null +++ b/config/superset/bootstrap_pipeline_dashboard.py @@ -0,0 +1,367 @@ +#!/usr/bin/env python3 +"""Create Superset dashboard for Debezium, Kafka CDC changes, and Spark.""" +import json +import os +import requests + +BASE = "http://127.0.0.1:8088" +DASH_TITLE = "ATC Lakehouse · Pipeline & CDC" +CSS_PATH = "/tmp/palantir_dashboard.css" + +MONITOR_URI = "trino://mo@10.0.21.50:8089/postgres_sales/monitor" +KAFKA_URI = "trino://mo@10.0.21.50:8089/kafka/default" + + +def session(): + s = requests.Session() + r = s.post( + f"{BASE}/api/v1/security/login", + json={"username": "admin", "password": "admin", "provider": "db", "refresh": True}, + ) + r.raise_for_status() + h = {"Authorization": "Bearer " + r.json()["access_token"], "Content-Type": "application/json"} + h["X-CSRFToken"] = s.get(f"{BASE}/api/v1/security/csrf_token/", headers=h).json()["result"] + h["Referer"] = BASE + return s, h + + +def get_db(s, h, name, uri): + r = s.get(f"{BASE}/api/v1/database/", headers=h) + for d in r.json().get("result", []): + if d["database_name"] == name: + return d["id"] + r = s.post( + f"{BASE}/api/v1/database/", + headers=h, + json={"database_name": name, "sqlalchemy_uri": uri, "expose_in_sqllab": True}, + ) + r.raise_for_status() + return r.json()["id"] + + +def ds_table(s, h, db_id, schema, table): + r = s.get(f"{BASE}/api/v1/dataset/", headers=h) + for d in r.json().get("result", []): + if d.get("table_name") == table and d.get("schema") == schema and d.get("database", {}).get("id") == db_id: + return d["id"] + r = s.post( + f"{BASE}/api/v1/dataset/", + headers=h, + json={"database": db_id, "schema": schema, "table_name": table}, + ) + r.raise_for_status() + return r.json()["id"] + + +def ds_sql(s, h, db_id, name, sql): + r = s.get(f"{BASE}/api/v1/dataset/", headers=h) + for d in r.json().get("result", []): + if d.get("table_name") == name: + return d["id"] + r = s.post( + f"{BASE}/api/v1/dataset/", + headers=h, + json={"database": db_id, "table_name": name, "sql": sql}, + ) + r.raise_for_status() + return r.json()["id"] + + +def chart(s, h, name, ds_id, viz, params): + r = s.get(f"{BASE}/api/v1/chart/", headers=h) + for c in r.json().get("result", []): + if c.get("slice_name") == name: + return c["id"] + p = {"datasource": f"{ds_id}__table", "viz_type": viz, "row_limit": 1000, **params} + r = s.post( + f"{BASE}/api/v1/chart/", + headers=h, + json={ + "slice_name": name, + "viz_type": viz, + "datasource_id": ds_id, + "datasource_type": "table", + "params": json.dumps(p), + "owners": [1, 2, 3], + }, + ) + r.raise_for_status() + return r.json()["id"] + + +def layout(items): + L = { + "DASHBOARD_VERSION": "v2", + "ROOT_ID": {"type": "ROOT", "id": "ROOT_ID", "children": ["GRID_ID"]}, + "GRID_ID": {"type": "GRID", "id": "GRID_ID", "children": [], "parents": ["ROOT_ID"]}, + } + ri = 0 + + def row(): + nonlocal ri + ri += 1 + rid = f"ROW-{ri}" + L["GRID_ID"]["children"].append(rid) + L[rid] = { + "type": "ROW", + "id": rid, + "children": [], + "parents": ["ROOT_ID", "GRID_ID"], + "meta": {"background": "BACKGROUND_TRANSPARENT"}, + } + return rid + + for item in items: + if item[0] == "md": + rid = row() + mid = f"MD-{ri}" + L[rid]["children"].append(mid) + L[mid] = { + "type": "MARKDOWN", + "id": mid, + "children": [], + "parents": ["ROOT_ID", "GRID_ID", rid], + "meta": {"width": 12, "height": 10, "code": f"## {item[1]}\n\n{item[2]}"}, + } + else: + cid, name = item + rid = row() + charts_in_row = [k for k in L[rid]["children"] if k.startswith("CHART-")] + if len(charts_in_row) >= 3: + rid = row() + key = f"CHART-{cid}" + L[rid]["children"].append(key) + hgt = 70 if "Recent" in name or "table" in name.lower() else 55 + L[key] = { + "type": "CHART", + "id": key, + "children": [], + "parents": ["ROOT_ID", "GRID_ID", rid], + "meta": {"width": 4 if "Recent" not in name else 12, "height": hgt, "chartId": cid, "sliceName": name}, + } + return L + + +def main(): + s, h = session() + db_mon = get_db(s, h, "Trino · Pipeline Monitor", MONITOR_URI) + db_kfk = get_db(s, h, "Trino · Kafka CDC", KAFKA_URI) + + ds_conn = ds_table(s, h, db_mon, "monitor", "debezium_connectors") + ds_topics = ds_table(s, h, db_mon, "monitor", "kafka_topics") + ds_ops = ds_table(s, h, db_mon, "monitor", "cdc_operations") + ds_recent = ds_table(s, h, db_mon, "monitor", "cdc_recent_events") + ds_spark = ds_table(s, h, db_mon, "monitor", "spark_applications") + + LIVE_CDC_SQL = """ +SELECT + 'PostgreSQL' AS source_system, + CASE json_extract_scalar(_message, '$.payload.op') + WHEN 'c' THEN 'INSERT' WHEN 'u' THEN 'UPDATE' WHEN 'd' THEN 'DELETE' WHEN 'r' THEN 'SNAPSHOT' ELSE 'OTHER' + END AS change_type, + json_extract_scalar(_message, '$.payload.source.table') AS table_name, + json_extract_scalar(_message, '$.payload.after.order_id') AS record_key, + _timestamp AS event_time +FROM kafka.default."postgres-sales.public.sales_orders" +WHERE _timestamp > current_timestamp - INTERVAL '7' DAY +LIMIT 500 +""" + ds_live = ds_sql(s, h, db_kfk, "live_cdc_postgres_sample", LIVE_CDC_SQL) + + charts = [] + charts.append(chart(s, h, "Debezium · Connector Status", ds_conn, "table", {})) + charts.append( + chart( + s, + h, + "Debezium · RUNNING vs State", + ds_conn, + "pie", + { + "metric": {"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "COUNT(*)"}, + "groupby": ["state"], + }, + ) + ) + charts.append( + chart( + s, + h, + "Kafka · Topic Offsets", + ds_topics, + "echarts_timeseries_bar", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "SUM(end_offset)", "label": "Messages"} + ], + "groupby": ["topic"], + "row_limit": 20, + }, + ) + ) + charts.append( + chart( + s, + h, + "Kafka · Partitions per Topic", + ds_topics, + "echarts_timeseries_bar", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "SUM(end_offset)", "label": "Offset"} + ], + "groupby": ["topic", "partition_id"], + "row_limit": 30, + }, + ) + ) + charts.append( + chart( + s, + h, + "CDC · Changes by Type", + ds_ops, + "pie", + { + "metric": {"expressionType": "SQL", "sqlExpression": "SUM(event_count)", "label": "Events"}, + "groupby": ["operation_label"], + }, + ) + ) + charts.append( + chart( + s, + h, + "CDC · Changes per Source", + ds_ops, + "echarts_timeseries_bar", + { + "metrics": [ + {"expressionType": "SQL", "sqlExpression": "SUM(event_count)", "label": "Events"} + ], + "groupby": ["source_system", "operation_label"], + "row_limit": 20, + }, + ) + ) + charts.append( + chart( + s, + h, + "CDC · Recent Changes (sampled)", + ds_recent, + "table", + { + "all_columns": [ + "source_system", + "operation_label", + "table_name", + "record_key", + "detail", + "event_ts", + ], + "row_limit": 50, + }, + ) + ) + charts.append( + chart( + s, + h, + "CDC · Live Stream Sample (PostgreSQL)", + ds_live, + "table", + { + "all_columns": ["source_system", "change_type", "table_name", "record_key", "event_time"], + "row_limit": 100, + }, + ) + ) + charts.append( + chart( + s, + h, + "Spark · Applications", + ds_spark, + "table", + {"all_columns": ["app_id", "app_name", "state", "cores", "memory_mb", "duration_sec"]}, + ) + ) + charts.append( + chart( + s, + h, + "Pipeline · Total Kafka Messages", + ds_topics, + "big_number_total", + { + "metric": { + "expressionType": "SQL", + "sqlExpression": "SUM(end_offset)", + "label": "Total Offset", + } + }, + ) + ) + + chart_specs = [ + ("md", "Pipeline & Change Data Capture", "Debezium → Kafka → Spark · Live CDC visibility"), + ("md", "Debezium Connect", "Connector health on kafka01 :8083"), + (charts[0], "Debezium · Connector Status"), + (charts[1], "Debezium · RUNNING vs State"), + ("md", "Apache Kafka", "Topic volume & CDC streams on kafka01"), + (charts[2], "Kafka · Topic Offsets"), + (charts[3], "Kafka · Partitions per Topic"), + (charts[9], "Pipeline · Total Kafka Messages"), + ("md", "Data Changes (CDC)", "INSERT / UPDATE / DELETE / SNAPSHOT — gewijzigde data"), + (charts[4], "CDC · Changes by Type"), + (charts[5], "CDC · Changes per Source"), + (charts[6], "CDC · Recent Changes (sampled)"), + (charts[7], "CDC · Live Stream Sample (PostgreSQL)"), + ("md", "Apache Spark", "Batch & streaming jobs · lake01:8080"), + (charts[8], "Spark · Applications"), + ] + items = chart_specs + cids = [c[0] for c in chart_specs if c[0] != "md"] + css = open(CSS_PATH).read() if os.path.exists(CSS_PATH) else "" + + payload = { + "dashboard_title": DASH_TITLE, + "published": True, + "position_json": json.dumps(layout(items)), + "css": css, + "json_metadata": json.dumps( + { + "color_scheme": "palantir_ops", + "refresh_frequency": 120, + "chart_configuration": { + str(c): {"id": c, "crossFilters": {"scope": "global", "chartsInScope": cids}} + for c in cids + }, + } + ), + "owners": [1, 2, 3], + } + + r = s.get(f"{BASE}/api/v1/dashboard/", headers=h) + dash_id = None + for d in r.json().get("result", []): + if d.get("dashboard_title") == DASH_TITLE: + dash_id = d["id"] + break + if dash_id: + r = s.put(f"{BASE}/api/v1/dashboard/{dash_id}", headers=h, json=payload) + else: + r = s.post(f"{BASE}/api/v1/dashboard/", headers=h, json=payload) + dash_id = r.json()["id"] + print("Dashboard", dash_id, r.status_code) + for cid in cids: + s.put(f"{BASE}/api/v1/chart/{cid}", headers=h, json={"dashboards": [dash_id], "owners": [1, 2, 3]}) + + # query contexts + os.system("python3 /tmp/fix_charts_qc.py 2>/dev/null || true") + print(f"URL: {BASE}/superset/dashboard/{dash_id}/") + + +if __name__ == "__main__": + main() diff --git a/config/superset/collect_pipeline_metrics.py b/config/superset/collect_pipeline_metrics.py new file mode 100644 index 0000000..4f81df6 --- /dev/null +++ b/config/superset/collect_pipeline_metrics.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Collect Debezium, Kafka CDC, and Spark metrics into postgres monitor schema.""" +import json +import shlex +import subprocess +import urllib.request +from collections import Counter +from datetime import datetime, timezone + +import psycopg2 + +PG_DSN = "host=10.0.21.51 dbname=postgres user=mo password=Dell2026!" +KAFKA = "10.0.21.36:9092" +DEBEZIUM = "http://localhost:8083" # Kafka Connect on kafka01; fallback lake01 :8083 +SPARK_MASTER = "http://10.0.21.50:8080" + +CDC_TOPICS = [ + ("PostgreSQL", "postgres-sales.public.sales_orders"), + ("MongoDB", "mongodb-supplychain.supplychain.events"), +] + +OP_LABELS = {"c": "INSERT", "u": "UPDATE", "d": "DELETE", "r": "SNAPSHOT", "i": "INSERT"} + + +def fetch_json(url, timeout=10): + with urllib.request.urlopen(url, timeout=timeout) as r: + return json.loads(r.read().decode()) + + +def collect_debezium(cur): + connectors = fetch_json(f"{DEBEZIUM}/connectors") + cur.execute("DELETE FROM monitor.debezium_connectors") + now = datetime.now(timezone.utc) + for name in connectors: + try: + st = fetch_json(f"{DEBEZIUM}/connectors/{name}/status") + except Exception as e: + cur.execute( + """INSERT INTO monitor.debezium_connectors + (connector_name, state, task_state, worker_id, checked_at) + VALUES (%s,%s,%s,%s,%s)""", + (name, "ERROR", str(e)[:32], "", now), + ) + continue + conn_state = st.get("connector", {}).get("state", "UNKNOWN") + tasks = st.get("tasks") or [] + task_state = tasks[0].get("state", "NONE") if tasks else "NONE" + worker = st.get("connector", {}).get("worker_id", "") + cur.execute( + """INSERT INTO monitor.debezium_connectors + (connector_name, state, task_state, worker_id, checked_at) + VALUES (%s,%s,%s,%s,%s)""", + (name, conn_state, task_state, worker, now), + ) + + +KAFKA_BIN = "/opt/kafka/bin" +USE_SSH_KAFKA = False # set True when running off-host + + +def _kafka_cmd(bin_name, args): + parts = [f"{KAFKA_BIN}/{bin_name}"] + list(args) + if USE_SSH_KAFKA: + remote = " ".join(shlex.quote(p) for p in parts) + full = f"ssh -o StrictHostKeyChecking=no root@10.0.21.36 {remote}" + return subprocess.check_output(full, shell=True, stderr=subprocess.DEVNULL, timeout=90, text=True) + return subprocess.check_output(parts, stderr=subprocess.DEVNULL, timeout=90, text=True) + + +def kafka_end_offsets(topic): + try: + out = _kafka_cmd( + "kafka-run-class.sh", + [ + "kafka.tools.GetOffsetShell", + "--broker-list", + "localhost:9092", + "--topic", + topic, + ], + ) + except Exception: + return [] + rows = [] + for line in out.strip().splitlines(): + parts = line.split(":") + if len(parts) >= 3: + rows.append((int(parts[1]), int(parts[2]))) + return rows + + +def sample_topic_messages(topic, max_msgs=3000, tail=5000): + """Sample recent messages using kafka-console-consumer from tail.""" + offsets = kafka_end_offsets(topic) + if not offsets: + return [] + # Pick partition 0 for sampling + part, end = offsets[0] + start = max(0, end - tail) + try: + out = _kafka_cmd( + "kafka-console-consumer.sh", + [ + "--bootstrap-server", + "localhost:9092", + "--topic", + topic, + "--partition", + str(part), + "--offset", + str(start), + "--max-messages", + str(min(max_msgs, tail)), + "--timeout-ms", + "15000", + ], + ) + except Exception: + return [] + return [ln for ln in out.strip().split("\n") if ln.strip()] + + +def parse_debezium_line(line): + try: + doc = json.loads(line) + payload = doc.get("payload") or doc + op = payload.get("op") or payload.get("operationType") or "?" + src = payload.get("source") or {} + table = src.get("table") or src.get("collection") or "" + ts_ms = payload.get("ts_ms") or src.get("ts_ms") + after = payload.get("after") or {} + before = payload.get("before") or {} + row = after if after else before + key = str(row.get("order_id") or row.get("event_id") or row.get("_id") or "")[:200] + detail = str(row.get("region") or row.get("type") or row.get("department") or "")[:200] + event_ts = None + if ts_ms: + event_ts = datetime.fromtimestamp(int(ts_ms) / 1000, tz=timezone.utc) + return op, table, key, detail, event_ts + except Exception: + return None + + +def collect_kafka_cdc(cur): + now = datetime.now(timezone.utc) + cur.execute("DELETE FROM monitor.kafka_topics") + cur.execute("DELETE FROM monitor.cdc_operations") + cur.execute("DELETE FROM monitor.cdc_recent_events") + + for source, topic in CDC_TOPICS: + for part, end in kafka_end_offsets(topic): + cur.execute( + """INSERT INTO monitor.kafka_topics (topic, partition_id, end_offset, checked_at) + VALUES (%s,%s,%s,%s)""", + (topic, part, end, now), + ) + + lines = sample_topic_messages(topic, max_msgs=2000, tail=3000) + ops = Counter() + recent = [] + for line in lines: + parsed = parse_debezium_line(line) + if not parsed: + continue + op, table, key, detail, event_ts = parsed + ops[op] += 1 + if len(recent) < 100: + recent.append((op, table, key, detail, event_ts)) + + for op, cnt in ops.items(): + cur.execute( + """INSERT INTO monitor.cdc_operations + (source_system, topic, operation, operation_label, event_count, checked_at) + VALUES (%s,%s,%s,%s,%s,%s)""", + (source, topic, op, OP_LABELS.get(op, op), cnt, now), + ) + + for op, table, key, detail, event_ts in recent[:50]: + cur.execute( + """INSERT INTO monitor.cdc_recent_events + (source_system, topic, operation, operation_label, table_name, + record_key, detail, event_ts, sampled_at) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)""", + ( + source, + topic, + op, + OP_LABELS.get(op, op), + table, + key, + detail, + event_ts, + now, + ), + ) + + +def collect_spark(cur): + now = datetime.now(timezone.utc) + cur.execute("DELETE FROM monitor.spark_applications") + try: + data = fetch_json(f"{SPARK_MASTER}/json/", timeout=5) + apps = [] + if isinstance(data, dict): + # Standalone master JSON + for a in data.get("activeapps", []) or []: + apps.append(a) + for a in data.get("completedapps", []) or []: + apps.append(a) + for a in apps[:20]: + cur.execute( + """INSERT INTO monitor.spark_applications + (app_id, app_name, state, cores, memory_mb, duration_sec, checked_at) + VALUES (%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (app_id) DO UPDATE SET + app_name=EXCLUDED.app_name, state=EXCLUDED.state, + cores=EXCLUDED.cores, memory_mb=EXCLUDED.memory_mb, + duration_sec=EXCLUDED.duration_sec, checked_at=EXCLUDED.checked_at""", + ( + a.get("id", "unknown"), + a.get("name", "Spark App"), + "RUNNING" if "attempts" not in a else "COMPLETED", + int(a.get("cores", 0) or 0), + int((a.get("memory", 0) or 0) / 1024 / 1024), + int(a.get("duration", 0) / 1000) if a.get("duration") else 0, + now, + ), + ) + except Exception as e: + # Placeholder row so dashboard shows Spark host status + cur.execute( + """INSERT INTO monitor.spark_applications + (app_id, app_name, state, cores, memory_mb, duration_sec, checked_at) + VALUES (%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (app_id) DO UPDATE SET state=EXCLUDED.state, checked_at=EXCLUDED.checked_at""", + ( + "spark-master", + f"Spark Master @ {SPARK_MASTER}", + "REACHABLE" if "Connection" not in str(e) else "UNREACHABLE", + 0, + 0, + 0, + now, + ), + ) + + +def main(): + conn = psycopg2.connect(PG_DSN) + conn.autocommit = True + cur = conn.cursor() + print("Collecting Debezium...") + collect_debezium(cur) + print("Collecting Kafka CDC samples...") + collect_kafka_cdc(cur) + print("Collecting Spark...") + collect_spark(cur) + cur.close() + conn.close() + print("Done.") + + +if __name__ == "__main__": + main() diff --git a/config/superset/fix_charts_qc.py b/config/superset/fix_charts_qc.py new file mode 100644 index 0000000..b059edd --- /dev/null +++ b/config/superset/fix_charts_qc.py @@ -0,0 +1,85 @@ +"""Generate and save query_context for API-created Superset charts.""" +import json + +app = __import__("superset.app", fromlist=["create_app"]).create_app() + +with app.app_context(): + from flask import g + from superset.extensions import db + from superset.models.slice import Slice + from superset.models.core import Database + from superset.charts.schemas import ChartDataQueryContextSchema + from superset import security_manager + + admin = security_manager.find_user(username="admin") + g.user = admin + + charts = db.session.query(Slice).order_by(Slice.id).all() + for sl in charts: + try: + fd = sl.form_data + metric = fd.get("metric") + metrics = fd.get("metrics") or ([metric] if metric else []) + if not metrics: + metrics = [ + { + "expressionType": "SQL", + "sqlExpression": "COUNT(*)", + "label": "COUNT(*)", + } + ] + + groupby = fd.get("groupby") or [] + payload = { + "datasource": { + "id": sl.datasource_id, + "type": sl.datasource_type, + }, + "force": False, + "queries": [ + { + "filters": [], + "extras": {"having": "", "where": ""}, + "applied_time_extras": {}, + "columns": groupby if isinstance(groupby, list) else [], + "metrics": metrics, + "orderby": [], + "annotation_layers": [], + "row_limit": int(fd.get("row_limit") or 1000), + "series_limit": 0, + "order_desc": True, + "url_params": {}, + "custom_params": {}, + "custom_form_data": {}, + } + ], + "form_data": fd, + "result_format": "json", + "result_type": "full", + } + qc = ChartDataQueryContextSchema().load(payload) + ctx = qc.cache_values if hasattr(qc, "cache_values") else None + if ctx is None: + # fallback: store factory input dict + from superset.common.query_context_factory import QueryContextFactory + + factory = QueryContextFactory() + ctx = { + "datasource": { + "id": sl.datasource_id, + "type": sl.datasource_type, + }, + "force": False, + "queries": payload["queries"], + "form_data": fd, + "result_format": "json", + "result_type": "full", + } + sl.query_context = json.dumps(ctx) if isinstance(ctx, dict) else json.dumps(payload) + sl.query_context_generation = True + db.session.add(sl) + print("OK", sl.id, sl.slice_name[:50]) + except Exception as e: + print("ERR", sl.id, sl.slice_name[:40], e) + db.session.commit() + print("committed") diff --git a/config/superset/palantir_dashboard.css b/config/superset/palantir_dashboard.css new file mode 100644 index 0000000..7be867c --- /dev/null +++ b/config/superset/palantir_dashboard.css @@ -0,0 +1,97 @@ +/* ATC Lakehouse dashboard — Palantir OPS overlay */ +.dashboard-wrapper, +.dashboard, +.grid-container, +.dashboard-content, +.dashboard-component-chart-holder { + background: transparent !important; +} + +.dashboard-header-container { + background: linear-gradient(135deg, rgba(12, 28, 52, 0.95) 0%, rgba(6, 20, 40, 0.98) 100%) !important; + border-bottom: 1px solid rgba(56, 132, 220, 0.25) !important; + backdrop-filter: blur(12px); +} + +.dashboard-header .dashboard-title { + font-family: 'DM Sans', system-ui, sans-serif !important; + font-weight: 700 !important; + letter-spacing: -0.02em !important; + background: linear-gradient(90deg, #e8eef7 0%, #22d3ee 50%, #fb923c 100%); + -webkit-background-clip: text; + -webkit-text-fill-color: transparent; + background-clip: text; +} + +.dashboard-component { + background: rgba(12, 28, 52, 0.72) !important; + border: 1px solid rgba(56, 132, 220, 0.2) !important; + border-radius: 12px !important; + box-shadow: 0 4px 24px rgba(0, 0, 0, 0.35), inset 0 1px 0 rgba(255, 255, 255, 0.04) !important; + backdrop-filter: blur(8px); + transition: border-color 0.2s ease, box-shadow 0.2s ease; +} + +.dashboard-component:hover { + border-color: rgba(251, 146, 60, 0.35) !important; + box-shadow: 0 8px 32px rgba(37, 99, 235, 0.2) !important; +} + +.chart-header, +.header-title, +.header-line { + color: #e8eef7 !important; + font-family: 'DM Sans', system-ui, sans-serif !important; +} + +.slice_container, +.chart-container, +.dashboard-chart-id { + background: transparent !important; +} + +/* Big number / KPI tiles */ +.big-number .header-line, +.big-number-viz .header-line { + color: #22d3ee !important; + font-size: 2.5rem !important; + font-weight: 700 !important; + text-shadow: 0 0 24px rgba(34, 211, 238, 0.35); +} + +/* Markdown section headers */ +.dashboard-markdown, +.markdown-component { + background: linear-gradient(90deg, rgba(59, 130, 246, 0.12), transparent) !important; + border-left: 3px solid #3b82f6 !important; + padding: 12px 16px !important; + border-radius: 0 8px 8px 0 !important; +} + +.dashboard-markdown h1, +.dashboard-markdown h2, +.markdown-component h1, +.markdown-component h2 { + color: #e8eef7 !important; + font-family: 'DM Sans', sans-serif !important; + margin: 0 !important; +} + +.dashboard-markdown p, +.markdown-component p { + color: #94a3b8 !important; + margin: 4px 0 0 !important; +} + +/* Filter bar */ +.filter-status-pane, +.dashboard-filters-panel { + background: rgba(6, 20, 40, 0.9) !important; + border: 1px solid rgba(56, 132, 220, 0.2) !important; + border-radius: 10px !important; +} + +/* Grid subtle glow */ +.grid-row { + margin-bottom: 8px; +} diff --git a/config/superset/save_query_context.py b/config/superset/save_query_context.py new file mode 100644 index 0000000..922494d --- /dev/null +++ b/config/superset/save_query_context.py @@ -0,0 +1,15 @@ +app = __import__("superset.app", fromlist=["create_app"]).create_app() + +with app.app_context(): + from superset.extensions import db + from superset.models.slice import Slice + charts = db.session.query(Slice).all() + for sl in charts: + try: + sl.query_context = sl.get_query_context() + db.session.add(sl) + print("saved", sl.id, (sl.slice_name or "")[:50]) + except Exception as e: + print("err", sl.id, e) + db.session.commit() + print("committed", len(charts), "charts") diff --git a/config/superset/superset_config.py b/config/superset/superset_config.py index 0ebd1d2..697511b 100644 --- a/config/superset/superset_config.py +++ b/config/superset/superset_config.py @@ -1,39 +1,117 @@ import os -# Secret key for session signing -SECRET_KEY = os.environ.get('SUPERSET_SECRET_KEY', 'your-secret-key-here') +SECRET_KEY = os.environ.get("SUPERSET_SECRET_KEY", "your-secret-key-here") +SQLALCHEMY_DATABASE_URI = "sqlite:////app/superset_home/superset.db" -# Database configuration - use SQLite to avoid psycopg2 issues -SQLALCHEMY_DATABASE_URI = 'sqlite:////app/superset_home/superset.db' - -# Redis cache configuration CACHE_CONFIG = { - 'CACHE_TYPE': 'redis', - 'CACHE_REDIS_URL': 'redis://redis:6379/0', - 'CACHE_DEFAULT_TIMEOUT': 300 + "CACHE_TYPE": "redis", + "CACHE_REDIS_URL": "redis://redis:6379/0", + "CACHE_DEFAULT_TIMEOUT": 300, } -# Enable CSRF protection ENABLE_PROXY_FIX = True - -# Feature flags -FEATURE_FLAGS = { - 'ENABLE_TEMPLATE_PROCESSING': True, - 'ALERT_REPORTS': True, -} - -# Row limit +TIMEZONE = "Europe/Amsterdam" ROW_LIMIT = 50000 -# Viz types -VIZ_TYPE_DICT = { - 'table': {}, - 'dist_bar': {}, - 'line': {}, - 'area': {}, - 'pie': {}, - 'number': {}, +# Branding — logo must be same-origin (/static/...) for CSP (img-src 'self') +APP_NAME = "Dell" +APP_ICON = "/static/assets/images/dell-logo.svg" +LOGO_TARGET_PATH = "/superset/welcome/" +LOGO_TOOLTIP = "Dell · ATC Lakehouse" + +FEATURE_FLAGS = { + "ENABLE_TEMPLATE_PROCESSING": True, + "ALERT_REPORTS": True, + "DASHBOARD_NATIVE_FILTERS": True, + "DASHBOARD_CROSS_FILTERS": True, + "ENABLE_ADVANCED_DATA_TYPES": True, } -# Timezone -TIMEZONE = 'Europe/Amsterdam' +# Allow icons server if needed for other assets (optional) +TALISMAN_ENABLED = True +TALISMAN_CONFIG = { + "content_security_policy": { + "base-uri": ["'self'"], + "default-src": ["'self'"], + "img-src": [ + "'self'", + "blob:", + "data:", + "https://apachesuperset.gateway.scarf.sh", + "https://static.scarf.sh/", + "http://atc-docker01.dell-atc.lan:8080", + "https://atc-docker01.dell-atc.lan:8080", + ], + "worker-src": ["'self'", "blob:"], + "connect-src": ["'self'"], + "object-src": "'none'", + "style-src": ["'self'", "'unsafe-inline'"], + "font-src": ["'self'"], + "script-src": ["'self'", "'strict-dynamic'"], + }, + "content_security_policy_nonce_in": ["script-src"], + "force_https": False, + "frame_options": "SAMEORIGIN", +} + +EXTRA_CATEGORICAL_COLOR_SCHEMES = [ + { + "id": "palantir_ops", + "description": "Palantir OPS — blue, cyan, orange, teal", + "label_colors": {}, + "isDefault": True, + "colors": [ + "#3b82f6", "#22d3ee", "#fb923c", "#2dd4bf", "#fbbf24", + "#a78bfa", "#f472b6", "#34d399", "#60a5fa", "#94a3b8", + ], + }, +] + +EXTRA_SEQUENTIAL_COLOR_SCHEMES = [ + { + "id": "palantir_blue", + "description": "Palantir blue gradient", + "isDefault": True, + "colors": ["#040c18", "#0c1a30", "#1e40af", "#3b82f6", "#22d3ee", "#7dd3fc"], + }, +] + +PALANTIR_FONTS = [ + "https://fonts.googleapis.com/css2?family=DM+Sans:ital,opsz,wght@0,9..40,400;0,9..40,500;0,9..40,600;0,9..40,700&family=JetBrains+Mono:wght@400;500;600&display=swap", +] + +PALANTIR_TOKENS = { + "brandAppName": "Dell", + "brandLogoAlt": "Dell", + "brandLogoUrl": "/static/assets/images/dell-logo.svg", + "brandLogoMargin": "8px 12px 8px 0", + "brandLogoHref": "/", + "brandLogoHeight": "32px", + "colorPrimary": "#007DB8", + "colorLink": "#22d3ee", + "colorSuccess": "#2dd4bf", + "colorWarning": "#fbbf24", + "colorError": "#f87171", + "colorInfo": "#38bdf8", + "colorBgBase": "#040c18", + "colorBgLayout": "#061428", + "colorBgContainer": "#0c1a30", + "colorBgElevated": "#0f2444", + "colorBorder": "#1e3a5f", + "colorBorderSecondary": "rgba(56, 132, 220, 0.22)", + "colorText": "#e8eef7", + "colorTextSecondary": "#94a3b8", + "colorTextTertiary": "#64748b", + "fontUrls": PALANTIR_FONTS, + "fontFamily": "'DM Sans', Inter, Helvetica, Arial, sans-serif", + "fontFamilyCode": "'JetBrains Mono', 'IBM Plex Mono', monospace", + "borderRadius": 8, + "borderRadiusLG": 12, +} + +THEME_DEFAULT = { + "algorithm": "dark", + "token": PALANTIR_TOKENS, +} +THEME_DARK = None +ENABLE_UI_THEME_ADMINISTRATION = False diff --git a/config/trino/catalog/kafka.properties b/config/trino/catalog/kafka.properties new file mode 100644 index 0000000..283330f --- /dev/null +++ b/config/trino/catalog/kafka.properties @@ -0,0 +1,4 @@ +connector.name=kafka +kafka.nodes=10.0.21.36:9092 +kafka.table-names=postgres-sales.public.sales_orders,mongodb-supplychain.supplychain.events,schema-changes.hr +kafka.hide-internal-columns=false diff --git a/config/trino/catalog/mongodb_supplychain.properties b/config/trino/catalog/mongodb_supplychain.properties index bf279b4..63d7d37 100644 --- a/config/trino/catalog/mongodb_supplychain.properties +++ b/config/trino/catalog/mongodb_supplychain.properties @@ -1,3 +1,3 @@ connector.name=mongodb -mongodb.connection-url=mongodb://10.0.21.51:27017/ +mongodb.connection-url=mongodb://mo:Dell2026%21@10.0.21.51:27017/?authSource=admin mongodb.schema-collection=__trino_schema