Skip to content

Lessons: Airflow 3.x

Breaking changes, JWT auth, LocalExecutor pitfalls, DAG debugging, yt-dlp rate limiting.

Source: extracted from AGENTS.md

Lessons Learned: Deploying Airflow 3.x

Airflow 3.x Breaking Changes (vs 2.x)

Aspect Airflow 2.x Airflow 3.x
Auth FAB (airflow.api.auth.backend) Simple Auth Manager (JWT-based)
CLI users airflow users create Broken (FAB provider bug) — dùng SQLAlchemy trực tiếp
DB init airflow db init airflow db migrate
PythonOperator airflow.operators.python airflow.providers.standard.operators.python
API auth Basic Auth / session JWT only — curl -u KHÔNG hoạt động

JWT Auth Flow (Simple Auth Manager)

  1. GET /auth/token/login → 307 redirect, set _token cookie (JWT)
  2. Extract token từ Set-Cookie header
  3. Authorization: Bearer <token> cho tất cả API calls
  4. all_admins=True → anonymous admin token, KHÔNG cần username/password
  5. Token endpoint KHÔNG phải /api/v2/auth/token — nó ở /auth/token/login (sub-app mount)

Ba Config Bắt Buộc (LocalExecutor)

# Thiếu bất kỳ cái nào = task execution fail
AIRFLOW__API__BASE_URL: http://<apiserver-container>:8080 # SDK client connects to API server
AIRFLOW__API_AUTH__JWT_SECRET: <fixed-32byte-secret> # Tất cả processes phải dùng cùng key
AIRFLOW__CORE__FERNET_KEY: <fernet-key> # Encrypt connections/passwords trong DB

Symptoms khi thiếu:

Missing config Symptom
API__BASE_URL httpx.ConnectError: Connection refused — task subprocess fallback localhost:8080
API_AUTH__JWT_SECRET Signature verification failed — mỗi process generate random key
FERNET_KEY (changed) InvalidToken — encrypted data trong DB không decode được

Volume Ownership

  • Airflow chạy UID 99:100 (user: "99:100" trong compose — match Unraid nobody:users)
  • Host mount dirs phải chown -R 99:100 /mnt/user/appdata/airflow/
  • root:root → Permission denied khi scheduler write logs, mc write config
  • Airflow binary shebang broken cho UID 99 — dùng python3 -m airflow thay vì airflow trực tiếp
  • getpass.getuser() fails cho UID 99 — mc config dir phải tạo thủ công: mkdir -p /tmp/mc-config && echo "anonymous" > /tmp/mc-config/.anonymous

Connection Management

  • Connection credentials PHẢI match actual DB user
  • airflow connections add --conn-login airflow nhưng PostgreSQL chỉ có user postgres → auth failed
  • Luôn verify: psql -h <host> -U <user> -d <db> trước khi tạo connection
  • Corrupted connections (Fernet key mismatch): delete qua SQLAlchemy, KHÔNG qua CLI

DAG Debugging Checklist

  1. docker exec airflow-scheduler airflow dags reserialize — force re-parse
  2. docker logs airflow-scheduler --since 5m — check task execution errors
  3. Task logs: docker exec airflow-scheduler cat /opt/airflow/logs/dag_id=.../task_id=.../attempt=1.log
  4. up_for_retry ≠ crash — task executed nhưng logic fail (SQL error, wrong creds, network)
  5. API task log endpoint có thể trả "Could not read served logs" — log server chưa configured
  6. No log file = task never executed — check logical_date (must be in past) và scheduler "Logical date is in future" errors
  7. "Adopting or resetting orphaned tasks" lặp lại = XCom deadlock hoặc version mismatch

Useful Commands

# Get JWT token
TOKEN=$(curl -sv http://<host>:<port>/auth/token/login 2>&1 | grep -oP '_token=\K[^;]+')

# API calls
curl -H "Authorization: Bearer $TOKEN" http://<host>:<port>/api/v2/dags
curl -H "Authorization: Bearer $TOKEN" http://<host>:<port>/api/v2/connections

# Trigger DAG (logical_date MUST be in past!)
curl -X POST -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"logical_date": "2026-05-02T12:00:00Z"}' \
  http://<host>:<port>/api/v2/dags/<dag_id>/dagRuns

# Force-fail stuck run (unblocks max_active_runs=1)
curl -X PATCH -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"state": "failed"}' \
  http://<host>:<port>/api/v2/dags/<dag_id>/dagRuns/<run_id>

# Reserialize DAGs
docker exec airflow-scheduler airflow dags reserialize

# Delete corrupted connections via SQLAlchemy
docker exec airflow-scheduler python3 -c "
from airflow import settings; from sqlalchemy import text
s = settings.Session(); s.execute(text('DELETE FROM connection')); s.commit()
"

# Fix volume ownership (UID 99:100 for Unraid)
docker run --rm -v /mnt/user/appdata/airflow:/data alpine chown -R 99:100 /data

Komodo-Specific Airflow Lessons

logical_date MUST Be In the Past

  • Airflow 3.x scheduler từ chối dispatch tasks khi logical_date > current time
  • Task instances được tạo nhưng không bao giờ execute → sau ~5 phút scheduler tự mark run failed
  • Scheduler log: Logical date is in future: <date> (logged as ERROR, mỗi giây)
  • Fix: Khi trigger manual run qua API, luôn dùng logical_date ở quá khứ: 
    # CORRECT — past date
curl -X POST ... -d '{"logical_date": "2026-05-10T07:05:00+00:00"}'
# WRONG — future or current time
curl -X POST ... -d '{"logical_date": "2026-05-10T07:11:00+00:00"}'
  • Scheduled runs (schedule="0 18 * * *") tự động set logical_date ở past → không bị ảnh hưởng

XCom Deadlock with LocalExecutor (CRITICAL)

  • LocalExecutor fork() + large XCom payload = deadlock
  • Task push XCom thành công, nhưng downstream task không bao giờ được dispatch
  • Scheduler log: "Adopting or resetting orphaned tasks" lặp lại mỗi 5 phút
  • Fix: Merge discover + download vào single task — không truyền large data qua XCom
  • Rule: Với LocalExecutor, giữ XCom payload < 1KB. Dùng DB/files cho large data transfer giữa tasks

Task Execution Model (Airflow 3.x LocalExecutor)

  • Tasks run via API server (/execution/task-instances/{id}/run), KHÔNG phải direct fork
  • airflow tasks run command KHÔNG tồn tại trong 3.x — chỉ có test, clear, state
  • airflow tasks test chạy độc lập (không qua scheduler) — useful cho debugging
  • max_active_runs=1 — stuck runs block tất cả others; dùng PATCH API để force-fail: 
    curl -X PATCH -H "Authorization: Bearer $TOKEN" \
  "http://host:port/api/v2/dags/<dag_id>/dagRuns/<run_id>" \
  -H "Content-Type: application/json" \
  -d '{"state": "failed"}'

yt-dlp 429 Rate Limiting

  • YouTube 429 rate limit hits subtitle downloads; audio downloads thường unaffected
  • --ignore-errors + subprocess.run() (no return code check) — yt-dlp returns non-zero khi subtitle 429, nhưng audio vẫn OK
  • --sleep-interval 2 giữa videos + --sleep-requests 1 giữa HTTP requests giúp giảm rate limiting
  • DAG design: Check audio file existence thay vì yt-dlp return code

GitDagBundle Version Management

  • airflow dags reserialize chỉ re-parses existing versions — KHÔNG tạo new version từ tracking_repo
  • Sau git push, scheduler tự pull tracking_repo + tạo new version trong versions/
  • Nếu run reference một version chưa tồn tại → task fail (DAG code not found)
  • Fix: Copy tracking_repo to versions dir: cp -r tracking_repo/ versions/{commit_hash}/