Airflow

Dag Bundle이란, 왜 필요한가

Dag Bundle은 하나 이상의 Dag 파일과 그에 딸린 리소스(다른 파이썬 스크립트, 설정 파일 등)의 묶음이다. 로컬 디렉터리, Git 저장소, 또는 그 밖의 외부 시스템에서 Dag를 가져올 수 있고, 배포 관리자가 직접 커스텀 Bundle 클래스를 작성해 원하는 소스를 지원할 수도 있다. 하나의 Airflow 배포에 여러 개의 Bundle을 동시에 정의할 수 있어 Dag를 논리 단위로 조직화하기도 쉬워진다.

이는 Airflow 2 이전의 'Dags 폴더' 방식보다 강력하다. 예전에는 모든 Dag가 로컬 디스크의 한 곳에 있어야 했고, 그 Dag들을 그 위치에 가져다 놓는 일은 전적으로 배포 관리자의 책임이었다. Bundle은 Dag가 필요로 하는 모든 것을 상위 단위에서 버전 관리할 수 있게 해준다.

Bundle이 버저닝을 지원하면, 실행 도중 Dag가 갱신되더라도 하나의 Dag 실행(run) 전체가 같은 코드 버전을 사용하도록 보장할 수 있다. 이는 (1) 버전 관리 — Dag가 실행 중간에 갱신되어도 그 run은 시작 시점의 코드를 계속 사용, (2) 확장성 — 대량의 Dag를 논리 단위로 묶어 관리, (3) 유연성 — Git 등 외부 시스템과의 매끄러운 통합이라는 세 가지 이점을 제공한다.

Bundle 타입별 특성과 설정

Airflow는 네 가지 내장 Bundle 타입을 제공한다.

``LocalDagBundle`` (``airflow.dag_processing.bundles.local.LocalDagBundle``)은 로컬 디렉터리를 참조한다. 개발·테스트 환경에 적합하지만 버저닝을 지원하지 않아 태스크는 항상 디스크의 최신 코드로 실행된다. 기본으로 Airflow는 이 타입의 Bundle을 자동 추가하며, 유일한 kwarg인 ``path``는 지정하지 않으면 ``[core] dags_folder`` 설정값을 기본값으로 쓴다.

``GitDagBundle`` (``airflow.providers.git.bundles.git.GitDagBundle``)은 Git 저장소와 통합되며 버저닝을 지원하는 유일한 내장 타입이다. 필수 kwarg는 ``tracking_ref``(브랜치·태그·커밋 SHA)이고, 저장소 자격증명은 ``git_conn_id``로 Airflow Connection을 참조하거나 ``repo_url``을 직접 지정한다. ``subdir``로 체크아웃 범위를 서브디렉터리로 좁히거나, ``sparse_dirs``로 특정 디렉터리만 받는 sparse checkout도 가능하다. 각 Dag 실행은 자신이 생성될 때의 Git 커밋을 기록하므로, 저장소가 그 뒤 갱신되어도 재실행 시 동일 코드를 재사용할 수 있다.

``S3DagBundle``과 ``GCSDagBundle``은 각각 S3 버킷, GCS 버킷을 참조하며 둘 다 버저닝을 지원하지 않아 태스크는 항상 버킷의 최신 코드로 실행된다. 필수 kwarg는 ``bucket_name``이고, 각각 ``aws_conn_id``(기본값 ``aws_default``)/``gcp_conn_id``(기본값 ``google_cloud_default``)와 ``prefix``로 버킷 내 서브디렉터리를 좁힐 수 있다.

여러 Bundle 타입을 하나의 배포에 함께 쓸 수 있고, 기본으로 추가되는 ``LocalDagBundle``이 더 이상 필요 없다면 제거해도 된다. Bundle 설정은 ``[dag_processor] dag_bundle_config_list``에 ``name``·``classpath``·``kwargs``를 담은 딕셔너리들의 리스트(JSON 형식 문자열)로 지정한다.

뷰 URL(외부 시스템에서 Dag를 볼 수 있는 링크)을 커스터마이즈하고 싶다면 kwargs에 ``view_url_template``을 지정하면 된다. 이 템플릿은 ``{subdir}``처럼 Bundle 객체의 속성만 플레이스홀더로 쓸 수 있으며, 지정된 URL은 안전성 검증을 거쳐 안전하지 않다고 판단되면 뷰 URL이 ``None``으로 설정된다 — 안전하지 않은 URL로 인한 보안 문제를 막기 위함이다. Bundle별 새로고침 주기(``[dag_processor] refresh_interval``)도 kwargs로 개별 오버라이드할 수 있다.

Airflow 3.0.2부터는 base 이미지에 git이 사전 설치되어 있다. 그보다 이전 버전을 쓴다면 Dockerfile에 git을 직접 설치해야 한다.

자격증명 노출 위험, 사용자 impersonation

Bundle의 ``kwargs``는 ``[dag_processor] dag_bundle_config_list`` 설정에 저장되는데, ``[api] expose_config``가 활성화되어 있으면 Airflow는 이 설정을 Config API로 그대로 노출한다. 즉 설정을 읽을 권한이 있는 사용자라면 누구나 이 값을 그대로 읽을 수 있으므로, kwargs 안에 비밀 값을 넣어서는 안 된다.

예를 들어 ``repo_url``에 ``https://x-access-token:<token>@github.com/org/repo.git``처럼 토큰을 직접 박아 넣는 방식은 금지된다. 대신 Git은 ``git_conn_id``, S3는 ``aws_conn_id``, GCS는 ``gcp_conn_id``로 Airflow Connection을 참조하고, 실제 자격증명은 시크릿 백엔드에 보관해야 한다. Connection 필드는 런타임에 해석되며 ``dag_bundle_config_list``에는 기록되지 않는다.

``run_as_user``(사용자 impersonation)와 Bundle을 함께 쓸 때는, impersonate된 사용자가 메인 Airflow 프로세스가 만든 Bundle 파일에 접근할 수 있도록 파일 권한을 신경 써야 한다 — 모든 impersonate 대상 사용자와 Airflow 사용자를 같은 그룹에 두고, ``umask 0002`` 같은 적절한 umask를 설정하는 식이다. 다만 이 권한 기반 접근은 임시방편이며, 앞으로는 supervisor 기반 Bundle 연산을 통해 그룹 권한 공유 없이도 다중 사용자 접근을 처리할 계획이다.

재실행 시 버전 선택 — rerun_with_latest_version과 disable_bundle_versioning

사용자가 DAG 실행이나 태스크 인스턴스를 clear할 때, UI는 '최신 Bundle 버전으로 재실행할지, 원래 실행에 쓰인 버전으로 재실행할지' 묻는 체크박스를 보여준다. ``rerun_with_latest_version`` 설정은 이 체크박스의 기본 상태를 결정해, 팀이 매번 수동으로 결정하지 않아도 되게 해준다. 같은 설정은 API/CLI로 backfill을 생성할 때의 기본 ``run_on_latest_version`` 동작도 함께 지배한다. 단, 이는 GitDagBundle 같은 버저닝 지원 Bundle 타입에만 적용되며, LocalDagBundle은 버저닝을 지원하지 않으므로 항상 최신 코드를 쓴다.

각 DAG는 파싱 버전(``DagModel.bundle_version``)을 가지며, Dag processor가 파일을 다시 파싱할 때마다 갱신된다. 각 DAG 실행은 자신이 생성될 때의 Bundle 버전을 기록한다. ``rerun_with_latest_version``이 ``False``면 DAG 실행을 clear해도 원래 Bundle 버전이 보존되어 같은 코드로 재실행되므로(실패 디버깅에 유리), ``True``면 clear 시 현재 파싱된 버전으로 갱신되어 최신 코드로 재실행된다.

이 설정은 다음 우선순위(높은 것부터)로 해석된다: (1) API 요청 바디의 ``run_on_latest_version`` 파라미터가 명시되어 있으면 그 값, (2) DAG 레벨의 ``rerun_with_latest_version`` 파라미터가 True/False로 설정되어 있으면 그 값, (3) 전역 설정 ``[core] rerun_with_latest_version``, (4) 어디에도 없으면 호출 경로별 과거 기본값(clear/rerun은 False, backfill은 True)을 적용한다. 예외적으로, 자신만의 버전이 없는 DAG 실행 — Airflow 2에서 이어져 온 것이거나 ``airflow db clean``으로 버전 기록이 지워진 경우 — 은 보존할 것이 없으므로 이 설정과 무관하게 항상 최신 버전으로 clear된다.

``disable_bundle_versioning``은 이와 별개의 설정으로, 버전 추적 자체를 완전히 끈다 — ``True``면 DAG 실행에 ``bundle_version``이 아예 기록되지 않는다. DAG 파라미터와 전역 설정(``[dag_processor] disable_bundle_versioning``) 양쪽으로 지정할 수 있다. 반면 ``rerun_with_latest_version``은 버전 추적은 계속 켜둔 채로 재실행 시 기본 선택만 바꾼다. 즉 ``disable_bundle_versioning``은 '버전을 아예 추적할 것인가'에 답하고, ``rerun_with_latest_version``은 '재실행 시 어느 버전을 기본으로 고를 것인가'에 답한다 — 버저닝이 꺼져 있으면 ``rerun_with_latest_version``은 아무 효과가 없다.

커스텀 Dag Bundle 작성하기

``BaseDagBundle``을 확장해 자신만의 Bundle 클래스를 만들 때 반드시 구현해야 하는 추상 메서드가 세 가지 있다. ``path`` 프로퍼티는 이 Bundle의 Dag 파일이 저장된 디렉터리의 ``Path``를 반환해야 한다. ``get_current_version``은 Bundle의 현재 버전을 문자열로 반환해야 하며(버저닝을 지원하지 않으면 ``None``을 반환), Airflow는 이후 이 버전 값을 다시 ``__init__``에 전달해 해당 버전의 Bundle을 재현한다. ``refresh``는 Bundle 콘텐츠를 소스로부터 새로고침하는 로직(예: 원격 저장소에서 최신 변경분 pull)을 담당하며, Dag processor가 주기적으로 호출한다.

선택적으로 오버라이드할 수 있는 메서드도 있다. ``__init__``은 ``tracking_ref`` 같은 추가 파라미터로 Bundle을 초기화하도록 확장할 수 있지만, 부모 클래스의 ``__init__``을 반드시 호출해야 하고, 네트워크 호출 같은 비용이 큰 작업은 여기서 하지 말고 ``initialize`` 메서드로 미뤄야 한다(Bundle 인스턴스화 시점의 지연을 막기 위함). ``initialize``는 Dag processor나 worker에서 Bundle이 처음 쓰이기 직전에 호출되므로, 콘텐츠가 실제로 접근될 때만 비용이 큰 작업을 수행하도록 해준다. ``view_url``은 외부 시스템(예: Git 저장소의 웹 UI)에서 이 Bundle을 볼 수 있는 URL 문자열을 반환해야 한다.

버저닝을 지원하는 Bundle이라면 ``initialize``·``get_current_version``·``refresh``가 버전별 로직을 제대로 처리하도록 구현해야 한다.

동시성 측면에서, 여러 worker가 동시에 여러 Bundle 객체를 만들 수 있는데 Airflow는 Bundle 객체에 대한 호출을 직렬화해주지 않는다. 따라서 락이 필요한 기술(예: git repo clone)이라면 Bundle 클래스 자신이 락을 책임져야 한다 — 예를 들어 git repo를 클론하는 경우, 한 번에 하나의 Bundle 객체만 클론하도록 보장하는 책임은 Bundle 클래스에 있다. 이를 위해 base 클래스가 제공하는 ``lock`` 메서드를 활용할 수 있다.

Triggerer 컴포넌트에서는 Dag Bundle이 초기화되지 않는다는 제약도 있다. 실무적으로 이는 트리거가 Dag Bundle로부터 올 수 없다는 뜻이다 — Triggerer는 시간에 따른 트리거 코드 변경을 다루지 않고, 모든 것이 메인 프로세스에서 처리되기 때문이다. 커스텀 트리거를 쓰려면 Dag Bundle이 아니라 ``sys.path``상의 다른 곳에서 임포트 가능하도록 배치해야 한다.