Airflow 3 테스트·CI/CD·보안 실전 가이드

DAG를 잘 짜는 것과 그 DAG를 안전하고 반복 가능하게 운영하는 것은 다른 문제입니다. 손으로 파일을 서버에 복사해 넣던 시절이 지나면, 누군가는 반드시 "이 DAG, 배포 전에 검증했나요?", "워커가 메타DB 비밀번호를 들고 있나요?", "누가 이 파이프라인을 트리거할 수 있죠?" 같은 질문을 하게 됩니다. 이 글은 그 세 가지 질문 — 테스트·CI/CD·보안 — 에 Airflow 3 기준으로 답합니다.

이 글은 "Airflow 3 실전 연재"의 11편입니다. 직전 편 10편: 모니터링 & 운영에서 "돌아가는 파이프라인을 어떻게 들여다보는가"를 다뤘다면, 이번 편은 "어떻게 안전하게 거기까지 올리는가"를 다룹니다. 다음 편 12편: 프로덕션 베스트 프랙티스 체크리스트에서 전체를 한 장으로 묶습니다.

Airflow 3은 컴포넌트 구조가 바뀌면서 보안 모델도 함께 정돈됐습니다. 특히 DAG processor가 독립 프로세스로 분리되고, 워커가 더 이상 메타데이터 DB에 직접 붙지 않는다(Task Execution API)는 점은 보안 단원에서 중요하게 다룰 변화입니다. 자세한 아키텍처는 1편: 아키텍처 해부를 참고하세요.

1. DAG 테스트 — 무엇을, 어떤 순서로 검증하나

DAG 테스트는 거창한 인프라가 아니라, 세 개의 층으로 쌓아 올리는 일입니다. 아래로 갈수록 느리고 비싸지지만 확신은 커집니다.

1.1 import 오류 검사 — 가장 싸고 가장 많이 막아주는 그물

운영에서 DAG가 깨지는 가장 흔한 이유는 로직 버그가 아니라 임포트 실패입니다. 오타, 빠진 의존성, 잘못된 모듈 경로 하나면 DAG가 통째로 사라지죠. Airflow 3에서는 DAG processor가 파싱을 담당하므로, 파싱 실패는 곧 "그 DAG가 스케줄러 눈에 안 보임"을 뜻합니다.

CI에서 가장 먼저 돌려야 할 검사가 이것입니다. 모든 DAG 파일을 모아 DagBag으로 적재해 보고, import 에러가 하나라도 있으면 실패시키는 방식입니다.

# tests/test_dag_integrity.py
import pytest
from airflow.models import DagBag
 
 
@pytest.fixture(scope="session")
def dagbag():
    # include_examples=False: 번들된 예제 DAG는 검사 대상에서 제외
    return DagBag(dag_folder="dags/", include_examples=False)
 
 
def test_no_import_errors(dagbag):
    assert not dagbag.import_errors, (
        f"DAG import 오류:\n{dagbag.import_errors}"
    )
 
 
def test_dag_count(dagbag):
    # DAG가 통째로 사라지는 회귀를 막는 최소 가드
    assert len(dagbag.dags) >= 1

여기에 "모든 DAG에는 owner와 태그가 있어야 한다", "retry는 최소 1회" 같은 조직 규칙을 자동 검사로 추가하면, 리뷰어가 매번 눈으로 확인하던 것을 CI가 대신 잡아줍니다.

1.2 단위 테스트 — 태스크 "함수"를 직접 부른다

Airflow 3의 Task SDK(from airflow.sdk import dag, task)로 짠 TaskFlow 함수는 결국 평범한 파이썬 함수입니다. 비즈니스 로직을 Airflow와 분리해 두면, 스케줄러를 띄우지 않고도 그냥 호출해서 테스트할 수 있습니다.

# dags/sales_etl.py
from airflow.sdk import dag, task
 
 
def transform_rows(rows: list[dict]) -> list[dict]:
    # 순수 로직: Airflow를 전혀 모른다 → 테스트하기 쉽다
    return [r for r in rows if r["amount"] > 0]
 
 
@dag(schedule="@daily", catchup=False, tags=["sales"])
def sales_etl():
    @task
    def clean(rows: list[dict]) -> list[dict]:
        return transform_rows(rows)
 
    clean([])
 
 
sales_etl()

# tests/test_sales_etl.py
from dags.sales_etl import transform_rows
 
 
def test_transform_drops_non_positive():
    rows = [{"amount": 10}, {"amount": 0}, {"amount": -5}]
    assert transform_rows(rows) == [{"amount": 10}]

교훈: 로직은 Airflow 밖으로 빼라. @task 안에 핵심 계산을 욱여넣지 말고 순수 함수로 분리하면, 테스트가 빨라지고 재사용도 쉬워집니다.

1.3 통합 실행 — dag.test()와 airflow dags test

단위 테스트로 부품을 검증했다면, 다음은 DAG 한 회차가 실제로 끝까지 도는지를 봅니다. Airflow 3은 메타DB나 스케줄러 없이 단일 프로세스로 DAG 한 회를 실행하는 두 가지 길을 제공합니다.

# 파이썬 안에서: 디버거를 걸고 한 회차를 그대로 돌릴 수 있다
if __name__ == "__main__":
    sales_etl().test()

# CLI로: logical_date를 지정해 한 회차 실행
airflow dags test sales_etl 2026-07-04

dag.test()는 로컬 디버깅의 핵심 도구입니다. IDE 디버거를 붙여 태스크 내부를 한 줄씩 따라갈 수 있어, "실서버에 올려서 돌려봐야 안다"는 악순환을 끊어줍니다. 참고로 Airflow 3에서는 execution_date가 제거되어 logical_date 를 사용하며, asset/수동 트리거에서는 logical_date가 None일 수 있으니 시간 기반 로직은 data_interval_start/end를 쓰는 편이 안전합니다(자세한 내용은 6편: 스케줄링 & Asset 참고).

2. CI/CD 파이프라인 — commit에서 배포까지

테스트를 손으로 돌리면 결국 안 돌리게 됩니다. 모든 검사를 파이프라인에 박아 두고, 통과한 것만 배포되게 만드는 게 목적입니다. 단계는 단순한 직선입니다.

다음은 git push 한 번이 거치는 전형적인 흐름입니다. 앞 단계가 실패하면 뒤 단계는 시작조차 하지 않습니다.

핵심은 빠른 검사를 앞에, 느린 검사를 뒤에 두는 것입니다. 1초짜리 lint가 잡을 수 있는 문제를 5분짜리 이미지 빌드 뒤에서 발견하면 그만큼 낭비입니다.

GitHub Actions로 옮기면 대략 이런 모양입니다(예시).

# .github/workflows/airflow-ci.yml
name: airflow-ci
on: [push, pull_request]
 
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt apache-airflow
      - run: ruff check dags/                 # 1. lint
      - run: pytest tests/test_dag_integrity.py  # 2. import 테스트
      - run: pytest tests/                       # 3. 단위 테스트
 
  build:
    needs: test                # test가 통과해야만 실행
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: docker build -t registry.example.com/airflow:${{ github.sha }} .
      - run: docker push registry.example.com/airflow:${{ github.sha }}

2.1 DAG를 워커까지 전달하는 세 가지 방식

CI까지 통과했다고 끝이 아닙니다. 그 DAG 파일을 실제로 스케줄러와 워커가 보는 곳까지 가져다 놓는 단계, 즉 배포 전략이 남습니다. 크게 세 가지가 있고, 트레이드오프가 분명합니다.

Airflow 3에서 주목할 것은 DAG bundles입니다. "DAG가 어디서 오는가"(예: 특정 git 저장소·리비전)를 정의하는 메커니즘으로, DAG versioning 과 맞물려 어떤 실행이 어떤 버전의 DAG로 돌았는지를 UI에서 추적할 수 있게 해 줍니다. 과거 git-sync가 "그냥 최신 파일을 끌어오는" 방식이었다면, bundle은 출처와 버전을 1급 개념으로 끌어올린 셈입니다.

선택 기준: 재현성과 격리가 최우선이면 이미지에 굽기, 반복 속도가 최우선이면 git-sync/bundle. 의존성(파이썬 패키지)이 자주 바뀌면 어차피 이미지를 다시 빌드해야 하므로 "굽기"가 자연스럽습니다.

3. 보안 — 누가 무엇을, 어디서 할 수 있는가

보안은 "기능을 켠다/끈다"가 아니라 신뢰 경계를 긋는 일입니다. Airflow 3은 컴포넌트가 분리되면서 이 경계가 한결 또렷해졌습니다. 차례로 봅니다.

3.1 RBAC — 역할과 권한

Airflow는 역할 기반 접근 제어(RBAC) 를 제공합니다. 사용자에게 권한을 직접 주지 않고 역할(Role) 에 권한을 묶은 뒤, 사용자에게 역할을 부여하는 방식입니다. 기본 역할은 대략 이렇게 나뉩니다(예시).

핵심 원칙은 최소 권한입니다. 모두에게 Admin을 주면 RBAC를 켠 의미가 없습니다. 팀 단위로 커스텀 역할을 만들고, "이 팀은 자기 DAG만 본다" 같은 경계를 권한으로 표현하세요. 이는 뒤에서 다룰 멀티테넌시의 출발점이기도 합니다.

3.2 API server 인증 — JWT 토큰

Airflow 3에서 기존 webserver는 API server로 대체되었고, UI와 안정·버전드 REST API를 함께 제공합니다(구 /api/experimental은 제거됨). 이 REST API의 인증은 JWT 토큰 기반입니다. API server가 토큰을 발급하거나 외부 인증과 연동하고, 클라이언트는 매 요청에 그 토큰을 실어 보냅니다.

# 1) 토큰 발급 (예시 — 실제 경로/페이로드는 배포 구성에 따라 다름)
TOKEN=$(curl -s -X POST https://airflow.example.com/auth/token \
  -H "Content-Type: application/json" \
  -d '{"username":"ci-bot","password":"..."}' | jq -r .access_token)
 
# 2) 발급받은 JWT로 DAG 트리거
curl -X POST https://airflow.example.com/api/v2/dags/sales_etl/dagRuns \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"logical_date": "2026-07-04T00:00:00Z"}'

CI/CD에서 "배포 후 원격으로 스모크 트리거"를 걸 때 바로 이 흐름을 씁니다. REST API로 원격 스케줄을 다루는 더 자세한 내용은 9편: REST API & 원격 스케줄 변경을 참고하세요.

3.3 Secrets Backend — 비밀을 코드 밖으로

Connection·Variable에 든 비밀번호·토큰을 메타DB나 환경변수에 평문으로 두면 사고의 씨앗이 됩니다. Airflow는 Secrets Backend로 Vault·AWS Secrets Manager·GCP Secret Manager 같은 외부 비밀 저장소를 연결해, 비밀을 코드와 메타DB 밖에서 관리하도록 합니다. 설정과 패턴은 8편: 외부 시스템 연동 & 싱크 호출에서 다뤘으니 여기서는 "보안 관점에서 반드시 켜야 한다"는 점만 짚습니다.

3.4 DAG processor 분리와 Task Execution API — 격리의 핵심

여기가 Airflow 3 보안에서 가장 중요한 변화입니다. 두 가지가 맞물려 신뢰 경계를 새로 긋습니다.

• DAG processor 분리: DAG 파싱이 스케줄러에서 떨어져 나와 독립 프로세스로 돕니다. 사용자 코드(DAG 파일)를 실행하는 부분과 스케줄링 코어가 분리되어, 신뢰도가 다른 코드를 격리할 수 있습니다.
• Task Execution API: 워커(태스크)는 더 이상 메타데이터 DB에 직접 접속하지 않습니다. 대신 API server의 Task Execution Interface를 통해서만 상태를 주고받습니다.

이 변화의 보안 효과는 큽니다. 과거에는 모든 워커가 메타DB 자격증명을 들고 있어, 워커 한 대가 뚫리면 메타DB 전체가 노출되는 구조였습니다. Airflow 3에서는 워커가 메타DB를 모릅니다. 워커는 좁은 API 표면만 보고, 인증된 범위 안에서만 통신합니다. 덕분에 원격/엣지 워커도 안전하게 둘 수 있습니다(EdgeExecutor).

아래는 사용자 요청이 들어와 태스크가 실행되기까지의 신뢰 경계입니다. 점선이 보안 경계이고, 워커 쪽에서 메타DB로 향하는 직접 화살표가 없다는 점이 핵심입니다.

3.5 멀티테넌시 — "한 클러스터, 여러 팀"

여러 팀이 한 Airflow를 공유할 때는 위의 요소들을 조합해 경계를 만듭니다. Airflow는 단일 클러스터에서 OS 수준의 완벽한 테넌트 격리를 제공하지는 않으므로, 여러 겹의 방어로 접근합니다.

• RBAC 커스텀 역할로 팀별 DAG 접근을 제한한다.
• Pool과 priority_weight 로 자원을 분리해 한 팀이 슬롯을 독점하지 못하게 한다(3편: 환경설정 & 최적화 참고).
• Secrets Backend의 경로/정책으로 팀별 비밀 접근을 분리한다.
• 강한 격리가 필요하면 KubernetesExecutor로 태스크별 파드 격리를, 더 강하게는 팀별 클러스터 분리를 고려한다.

한 줄 정리: 멀티테넌시는 한 스위치가 아니라 RBAC + 자원 풀 + 비밀 분리 + 실행 격리를 겹쳐 쌓는 일입니다. 격리 요구가 아주 강하면 "나누는 것"(클러스터 분리)이 가장 단순한 답일 때도 많습니다.

4. 마무리 — 세 줄로 요약

• 테스트는 3층으로: import 검사(가장 싸다) → 단위 테스트(로직은 Airflow 밖으로) → dag.test()로 한 회차 통합 검증.
• CI/CD는 빠른 검사부터: lint → DAG import → 빌드 → 배포. 배포 방식은 재현성이면 이미지 굽기, 속도면 git-sync/DAG bundle.
• 보안은 경계 긋기: RBAC로 최소 권한, JWT로 API 인증, Secrets Backend로 비밀 격리. 그리고 Airflow 3의 가장 큰 선물 — 워커가 메타DB를 모른다는 사실을 활용하라.

다음 편 12편: 프로덕션 베스트 프랙티스 체크리스트에서는 지금까지 연재한 모든 내용을 배포 전 점검표 한 장으로 묶습니다. 거기서 만나요.

공식 문서: Apache Airflow Documentation