Files

399 lines
19 KiB
Bash
Raw Permalink Normal View History

2026-10-09 02:35:14 +07:00
#!/usr/bin/env bash
# =============================================================================
# eBdesk SiCABE — Verifikasi Phase 5 (Ingestion): Airflow
#
# Gate "kontrak ingestion": memastikan Airflow siap menjalankan pipeline —
# termasuk saat Airflow EXISTING (dikelola sistem lain).
#
# Yang diperiksa:
# A. Kontrak service : webserver + scheduler + dag-processor, health API
# (metadatabase & scheduler), airflow-init pernah jalan
# B. DAG : tanpa import error, sicabe_batch_daily ada & aktif,
# 4 task sesuai kontrak pipeline
# C. Connections : minio_s3 (endpoint/kredensial) DAN trino_default yang
# BENAR-BENAR DAPAT DIPAKAI (diuji konek nyata) — utang
# dari review Fase 3 (§7.6)
# D. Sumber & aset : sicabe_data.public.orders (sumber DAG), sling/env.yaml,
# DBT_TRINO_* untuk service Airflow
# E. Temuan/antarmuka: berkas yatim connections.sh, docstring DAG basi, dan
# validasi GE yang menunjuk tabel sumber → WARN; consumer
# fase 6 → SKIP
#
# Catatan: setiap koneksi diuji dengan PARAMETER TERSIMPAN-nya (bukan asumsi),
# dan kata sandi tidak pernah dicetak / tidak lewat argumen proses (via stdin).
#
# Exit code: 0 = semua gate lolos (WARN/SKIP boleh), 1 = ada FAIL.
#
# Usage:
# ./scripts/verify-phase5-ingestion.sh
# AIRFLOW_HOST=airflow.corp AIRFLOW_PORT=8080 ./scripts/verify-phase5-ingestion.sh
# =============================================================================
set -uo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
# shellcheck source=../lib/common.sh
source "${SCRIPT_DIR}/lib/common.sh"
# shellcheck source=../config/environment.defaults.sh
source "${SCRIPT_DIR}/config/environment.defaults.sh"
sicabe_load_env "${SCRIPT_DIR}" "$@"
# Config RELEVAN: mode host (--env-file) = BUNDLE host (ter-render), bukan deploy/config/ sumber.
if [[ -n "${SICABE_BUNDLE_DIR:-}" ]]; then CONFIG_DIR="${SICABE_BUNDLE_DIR}/config"; else CONFIG_DIR="${SCRIPT_DIR}/config"; fi
AIRFLOW_HOST="${AIRFLOW_HOST:-localhost}"
AIRFLOW_PORT="${AIRFLOW_PORT:-12082}"
AIRFLOW_URL="http://$(sicabe_local_host AIRFLOW_WEBSERVER_HOST AIRFLOW_HOST):${AIRFLOW_PORT}"
BATCH_DAG="${AIRFLOW_BATCH_DAG:-sicabe_batch_daily}"
CDC_DAG="${AIRFLOW_CDC_DAG:-sicabe_cdc_realtime}"
S3_ENDPOINT="${S3_ENDPOINT:-http://minio:9000}"
BUCKET="${MINIO_BUCKET_ICEBERG_DATA:-iceberg-data}"
EXPECTED_TASKS="bronze_to_silver ingest_to_bronze silver_to_gold validate_gold"
WEB_CONTAINER="sicabe-airflow-webserver"
AIRFLOW_EXTERNAL=0
if [[ "${AIRFLOW_HOST}" != "localhost" && "${AIRFLOW_HOST}" != "127.0.0.1" ]]; then
AIRFLOW_EXTERNAL=1
fi
PASS_N=0
FAIL_N=0
WARN_N=0
SKIP_N=0
pass() { echo -e " ${PASS} $1"; PASS_N=$((PASS_N + 1)); }
FAIL_MSGS=()
fail() { echo -e " ${FAIL} $1"; FAIL_N=$((FAIL_N + 1)); FAIL_MSGS+=("$1"); }
warn() { echo -e " ${WARN} $1"; WARN_N=$((WARN_N + 1)); }
skip() { echo -e " ${YELLOW}[SKIP]${NC} $1"; SKIP_N=$((SKIP_N + 1)); }
container_running() {
docker ps --format "{{.Names}}" 2>/dev/null | grep -q "^$1$"
}
af_cli() { docker exec "${WEB_CONTAINER}" airflow "$@" 2>/dev/null; }
# ─────────────────────────────────────────────────────────────────────────────
# A. Kontrak service
# ─────────────────────────────────────────────────────────────────────────────
check_service() {
echo -e "${BOLD}A. Kontrak service Airflow${NC}"
local c runtime=""
for c in "${WEB_CONTAINER}" sicabe-airflow-scheduler sicabe-airflow-dag-processor; do
if ! sicabe_host_expects_container "${c}"; then
skip "Container ${c} di host lain (mode host) - dilewati"
elif container_running "${c}"; then
runtime+=" ${c}"
fi
done
if [[ -n "${runtime}" ]]; then
pass "Container berjalan:${runtime}"
fi
if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
skip "Airflow dikelola eksternal (${AIRFLOW_HOST}:${AIRFLOW_PORT}) — uji container dilewati"
else
for c in "${WEB_CONTAINER}" sicabe-airflow-scheduler sicabe-airflow-dag-processor; do
if ! sicabe_host_expects_container "${c}"; then
skip "Container ${c} di host lain (mode host) - dilewati"
else
container_running "${c}" || fail "Container ${c} tidak berjalan"
fi
done
if docker ps -a --format "{{.Names}}" 2>/dev/null | grep -q "^sicabe-airflow-init$"; then
pass "Container one-shot sicabe-airflow-init ada (db migrate + users create)"
else
warn "Container sicabe-airflow-init tidak ditemukan — metadata DB mungkin belum di-migrate"
fi
fi
local health
health="$(curl -sf --max-time 15 "${AIRFLOW_URL}/api/v2/monitor/health" 2>/dev/null)"
if [[ -z "${health}" ]]; then
fail "Health API tidak menjawab di ${AIRFLOW_URL}/api/v2/monitor/health"
return 1
fi
local meta sched
meta="$(echo "${health}" | python3 -c "import json,sys;print(json.load(sys.stdin).get('metadatabase',{}).get('status',''))" 2>/dev/null)"
sched="$(echo "${health}" | python3 -c "import json,sys;print(json.load(sys.stdin).get('scheduler',{}).get('status',''))" 2>/dev/null)"
if [[ "${meta}" == "healthy" ]]; then
pass "metadatabase.status = healthy"
else
fail "metadatabase.status = '${meta:-tak terbaca}' (harus healthy)"
fi
if [[ "${sched}" == "healthy" ]]; then
pass "scheduler.status = healthy"
else
warn "scheduler.status = '${sched:-tak terbaca}'"
fi
}
# ─────────────────────────────────────────────────────────────────────────────
# B. DAG
# ─────────────────────────────────────────────────────────────────────────────
check_dags() {
echo -e "${BOLD}B. DAG pipeline${NC}"
if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
skip "Airflow eksternal — pemeriksaan DAG via CLI dilewati (gunakan UI/API di sana)"
return 0
fi
local errs
errs="$(af_cli dags list-import-errors)"
if [[ "${errs}" == *"No data found"* || -z "${errs}" ]]; then
pass "Tidak ada DAG import error"
else
fail "Ada DAG import error: $(echo "${errs}" | tr '\n' ' ' | cut -c1-120)"
fi
local list
list="$(af_cli dags list)"
if [[ "${list}" == *"${BATCH_DAG}"* ]]; then
pass "DAG '${BATCH_DAG}' ada"
else
fail "DAG '${BATCH_DAG}' tidak ditemukan — sudah disalin? (bootstrap.sh --step 6)"
fi
if [[ "${list}" == *"${CDC_DAG}"* ]]; then
pass "DAG '${CDC_DAG}' ada (opsional, default paused)"
else
warn "DAG '${CDC_DAG}' tidak ditemukan"
fi
# Baris DAG pada `dags list`: "<dag_id> | <fileloc> | <owner> | <is_paused> | ..."
local row
row="$(echo "${list}" | grep -m1 "^${BATCH_DAG} " || true)"
if [[ "${row}" == *"| False"* ]]; then
pass "DAG '${BATCH_DAG}' aktif (is_paused=False)"
else
# Gate ini SENGAJA read-only: ia melaporkan, tidak memperbaiki. Pengaktifan DAG
# dilakukan di bootstrap.sh step 6 (skrip yang memang memiliki DAG-nya), supaya
# hasil verifikasi tidak bergantung pada apakah gate pernah dijalankan.
warn "DAG '${BATCH_DAG}' paused — pipeline harian tidak akan jalan. Aktifkan: docker exec sicabe-airflow-webserver airflow dags unpause ${BATCH_DAG}"
fi
local tasks
tasks="$(af_cli tasks list "${BATCH_DAG}" | tr '\n' ' ' | tr -s ' ' | sed 's/ $//')"
local t missing=""
for t in ${EXPECTED_TASKS}; do
[[ " ${tasks} " == *" ${t} "* ]] || missing+=" ${t}"
done
if [[ -z "${missing}" ]]; then
pass "4 task sesuai kontrak: ${EXPECTED_TASKS}"
else
fail "Task hilang di '${BATCH_DAG}':${missing}"
fi
}
# ─────────────────────────────────────────────────────────────────────────────
# C. Connections — inti utang review Fase 3
# ─────────────────────────────────────────────────────────────────────────────
check_connections() {
echo -e "${BOLD}C. Connections (dipakai DAG user/operator)${NC}"
if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
skip "Airflow eksternal — pemeriksaan connection via CLI dilewati"
return 0
fi
local json
json="$(af_cli connections export -)"
if [[ -z "${json}" ]]; then
fail "Tidak bisa membaca connections (airflow connections export)"
return 1
fi
# ── minio_s3 ──────────────────────────────────────────────────────────────
local m_ep m_login expected_login
m_ep="$(echo "${json}" | python3 -c "
import json,sys
try:
c=json.load(sys.stdin)['minio_s3']; print(json.loads(c.get('extra') or '{}').get('endpoint_url',''))
except Exception: print('')" 2>/dev/null)"
m_login="$(echo "${json}" | python3 -c "
import json,sys
try: print(json.load(sys.stdin)['minio_s3'].get('login') or '')
except Exception: print('')" 2>/dev/null)"
if [[ "${json}" != *'"minio_s3"'* ]]; then
fail "Connection 'minio_s3' tidak ada (bootstrap.sh --step 6)"
elif [[ "${m_ep}" == "${S3_ENDPOINT}" ]]; then
pass "minio_s3: endpoint_url = ${m_ep}"
else
fail "minio_s3: endpoint_url '${m_ep:-kosong}' ≠ S3_ENDPOINT '${S3_ENDPOINT}'"
fi
# Kredensial harus mengikuti S3_ACCESS_KEY (fallback saat-pakai), bukan MINIO_ROOT_USER
# — kalau tidak, salah saat memakai S3 existing (§2.9).
expected_login="${S3_ACCESS_KEY:-${MINIO_ROOT_USER:-minioadmin}}"
if [[ -z "${m_login}" ]]; then
fail "minio_s3: login/kredensial kosong"
elif [[ "${m_login}" == "${expected_login}" ]]; then
pass "minio_s3: kredensial sesuai S3 aktif (login '${m_login}', password disamarkan)"
else
fail "minio_s3: login '${m_login}' ≠ kredensial S3 aktif '${expected_login}' — salah saat S3 existing"
warn " perbaiki: bootstrap.sh --step 6 (memakai \${S3_ACCESS_KEY} dengan fallback \${MINIO_ROOT_USER})"
fi
# ── trino_default — SENGAJA tidak disediakan (keputusan review Fase 5) ────
# Tidak ada DAG yang memakainya (task memakai dbt HTTPS 8443 + Sling), dan versi
# lamanya menunjuk http://trino:8080 tanpa kredensial → selalu 403.
if [[ "${json}" == *'"trino_default"'* ]]; then
local t_parsed
t_parsed="$(echo "${json}" | python3 -c "
import json,sys
try:
c=json.load(sys.stdin)['trino_default']
print('{0}:{1}'.format(c.get('host') or '', c.get('port') or ''))
except Exception: print('')" 2>/dev/null)"
fail "Connection 'trino_default' masih ada (${t_parsed}) — sudah diputuskan TIDAK disediakan"
warn " hapus: docker exec sicabe-airflow-webserver airflow connections delete trino_default"
else
pass "tidak ada connection 'trino_default' (sesuai keputusan — DAG memakai dbt + Sling)"
fi
}
# ─────────────────────────────────────────────────────────────────────────────
# D. Sumber data & aset pipeline
# ─────────────────────────────────────────────────────────────────────────────
check_source_and_assets() {
echo -e "${BOLD}D. Sumber data & aset pipeline${NC}"
# Sumber DAG: sicabe_data.public.orders (di-seed bootstrap step 6)
if ! sicabe_host_expects_container sicabe-postgres; then
skip "Container sicabe-postgres di host lain (mode host) - dilewati"
elif container_running sicabe-postgres; then
local n
n="$(docker exec -e PGPASSWORD="${POSTGRES_PASSWORD:-}" sicabe-postgres \
psql -U "${POSTGRES_USER:-sicabe_admin}" -d sicabe_data -tAc \
"SELECT count(*) FROM public.orders" 2>/dev/null || echo "")"
if [[ "${n}" =~ ^[0-9]+$ ]] && [[ ${n} -ge 1 ]]; then
pass "Sumber sicabe_data.public.orders → ${n} baris"
elif [[ -z "${n}" ]]; then
fail "sicabe_data.public.orders tidak ada / tidak terbaca (jalankan scripts/seed-source-postgres.sh)"
else
fail "sicabe_data.public.orders kosong (0 baris) — task ingest_to_bronze tidak akan menghasilkan data"
fi
else
skip "Container sicabe-postgres tidak berjalan — lewati cek sumber contoh"
fi
# Aset kredensial runtime Sling
local sling="${CONFIG_DIR}/airflow/sling/env.yaml"
if [[ -f "${sling}" ]]; then
pass "config/airflow/sling/env.yaml ada"
local key missing=""
for key in POSTGRES_DB ICEBERG rest_uri s3_endpoint; do
grep -q "${key}" "${sling}" || missing+=" ${key}"
done
if [[ -z "${missing}" ]]; then
pass " memuat koneksi POSTGRES_DB + ICEBERG (rest_uri & s3_endpoint)"
else
fail " env.yaml kehilangan:${missing} — generate ulang (bootstrap.sh / generate_runtime_credentials)"
fi
else
fail "config/airflow/sling/env.yaml tidak ada — task ingest_to_bronze akan gagal"
fi
# dbt → Trino: env DBT_TRINO_* harus SAMPAI ke container Airflow yang mengeksekusi
# task. Dicek dari ENV CONTAINER — bukan hasil render `docker compose config`.
# Alasan: (a) env container adalah yang benar-benar dipakai task dbt; (b) cek lama
# pernah salah lapor (19 PASS/1 FAIL, lalu 20/0 pada 3 pengulangan) sementara
# rendernya sendiri selalu memuat key itu → mekanisme render tidak dapat dipercaya
# sebagai bukti. `docker compose ... 2>/dev/null` juga menelan pesan error compose.
if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
skip "Airflow eksternal (${AIRFLOW_HOST}:${AIRFLOW_PORT}) — cek env DBT_TRINO_* dilewati"
elif ! container_running sicabe-airflow-scheduler; then
fail "Container sicabe-airflow-scheduler tidak berjalan — env DBT_TRINO_* tidak dapat diverifikasi"
else
local dbt_user dbt_pw
dbt_user="$(docker exec sicabe-airflow-scheduler printenv DBT_TRINO_USER 2>/dev/null || true)"
dbt_pw="$(docker exec sicabe-airflow-scheduler printenv DBT_TRINO_PASSWORD 2>/dev/null || true)"
if [[ -n "${dbt_user}" ]]; then
pass "Aset dbt: DBT_TRINO_USER='${dbt_user}' ada di env container Airflow"
else
fail "DBT_TRINO_USER tidak ada di env container Airflow — task dbt tidak bisa autentikasi"
fi
if [[ -n "${dbt_pw}" ]]; then
pass "Aset dbt: DBT_TRINO_PASSWORD ada di env container Airflow (nilai disamarkan)"
else
fail "DBT_TRINO_PASSWORD tidak ada di env container Airflow"
fi
fi
}
# ─────────────────────────────────────────────────────────────────────────────
# E. Temuan statis & antarmuka konsumen (WARN/SKIP)
# ─────────────────────────────────────────────────────────────────────────────
check_findings_and_consumers() {
echo -e "${BOLD}E. Temuan statis & antarmuka konsumen (informatif)${NC}"
local orphan="${CONFIG_DIR}/airflow/connections.sh"
if [[ -f "${orphan}" ]]; then
warn "config/airflow/connections.sh ADA tetapi tidak dirujuk siapa pun dan isinya menyimpang"
warn " (menyebut trino http:8080, postgres_default & openmetadata_default yang tidak pernah dibuat)"
else
pass "Tidak ada berkas connection yatim"
fi
local dag_file="${CONFIG_DIR}/airflow/dags/example_sicabe_pipeline.py"
if [[ -f "${dag_file}" ]]; then
if grep -q "postgres_default" "${dag_file}" 2>/dev/null; then
warn "Docstring DAG menyebut connection 'postgres_default' yang tidak dibuat bootstrap (DAG memakai env var)"
fi
if grep -q "trino-proxy" "${dag_file}" 2>/dev/null; then
warn "Komentar DAG masih menyebut 'trino-proxy' (service yang tidak ada; dbt konek langsung HTTPS 8443)"
fi
if grep -q "_postgres_conn_string" "${dag_file}" 2>/dev/null; then
warn "validate_gold/validate_bronze memvalidasi tabel di PostgreSQL (sicabe_data.orders = SUMBER),"
warn " bukan tabel gold/bronze di Iceberg — hasil 'LULUS' tidak membuktikan output pipeline"
fi
fi
if ! sicabe_host_expects_container sicabe-openmetadata; then
skip "Container sicabe-openmetadata di host lain (mode host) - dilewati"
elif container_running sicabe-openmetadata; then
pass "OpenMetadata berjalan — konektor Airflow diuji di gate Phase 6"
else
skip "OpenMetadata belum berjalan (fase 6) — lewati pemeriksaan antarmukanya"
fi
if ! sicabe_host_expects_container sicabe-grafana; then
skip "Container sicabe-grafana di host lain (mode host) - dilewati"
elif container_running sicabe-grafana; then
pass "Grafana berjalan — dashboard Airflow diuji di gate Phase 6"
else
skip "Grafana belum berjalan (fase 6) — lewati pemeriksaan antarmukanya"
fi
}
# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────
main() {
print_section "Verifikasi Phase 5 — Kontrak Ingestion (Airflow)"
echo -e " Airflow: ${CYAN}${AIRFLOW_URL}${NC} · DAG utama: ${CYAN}${BATCH_DAG}${NC}"
echo -e " Sumber : sicabe_data.public.orders (PostgreSQL) → Sling → dbt/Trino"
echo ""
check_service
check_dags
check_connections
check_source_and_assets
check_findings_and_consumers
echo ""
echo -e "${BOLD}Ringkasan:${NC} ${GREEN}${PASS_N} PASS${NC} · ${RED}${FAIL_N} FAIL${NC} · ${YELLOW}${WARN_N} WARN${NC} · ${YELLOW}${SKIP_N} SKIP${NC}"
if [[ ${FAIL_N} -gt 0 ]]; then
echo ""
echo -e "${BOLD}Yang gagal (${FAIL_N}):${NC}"
local _i=1 _m
for _m in "${FAIL_MSGS[@]}"; do
echo -e " ${RED}${_i}.${NC} ${_m}"
_i=$((_i + 1))
done
echo -e "${INFO} Perbaiki item di atas, lalu jalankan ulang: ./scripts/${0##*/}"
fi
if [[ ${FAIL_N} -eq 0 ]]; then
echo -e "${OK} ${GREEN}Kontrak ingestion Phase 5 terpenuhi — aman lanjut ke fase berikutnya.${NC}"
exit 0
fi
echo -e "${ERR} Ada ${FAIL_N} gate yang gagal — perbaiki sebelum melanjutkan."
exit 1
}
main "$@"