EcsRunTaskOperator에 skip_on_exit_code 옵션 추가 — Apache Airflow 기여 기록

관리비 및 공과금 데이터를 크롤링하는 파이프라인을 구축했습니다. 로컬 개발 환경에서는 크롤러를 Docker 컨테이너로 구동하여 정상 동작을 확인했고, 스케줄링은 Apache Airflow를 활용했습니다.

그러나 해당 크롤러를 dev 환경(AWS ECS)에 배포하여 실행하자, 로컬에서 정상 처리되던 태스크가 Airflow에서 실패(Failed) 상태로 종료되었습니다.

1. 현상: 로컬 Docker(스킵)와 dev ECS(실패)의 동작 차이

수도, 전기, 가스 관리비 도메인의 특성상, 고지서 발행일 특정일 기준 n영업일 이내라고 명시만 되어있고, 매번 동일한 일자에 고지가 되지 않습니다. 따라서 특정 월의 고지서가 아직 업로드되지 않는 케이스가 존재합니다. 이는 데이터를 수집할 수 없는 상태이므로, 태스크를 실패가 아닌 스킵(Skip)으로 처리해야 합니다.

이를 위해 크롤러가 데이터 미존재 시 0이 아닌 특정 exit code를 반환하며 종료되도록 설계했습니다. 그러나 같은 크롤러가 실행 환경에 따라 다르게 동작했습니다.

• 로컬 환경 (DockerOperator): 설정한 exit code가 Airflow에서 스킵(Skip) 상태로 정상 인식되었습니다.
• dev 환경 (EcsRunTaskOperator): 동일한 크롤러가 동일한 exit code를 반환했음에도 태스크가 실패(Failed)로 처리되었습니다.

EcsRunTaskOperator는 ECS 태스크의 exit code가 0이 아닐 경우, 예외 조건 없이 AirflowException을 발생시키며 태스크를 실패로 기록하고 있었습니다.

2. exit code의 정의와 활용

exit code(종료 코드)는 프로세스가 종료될 때 부모 프로세스에 전달하는 0~255 사이의 정수 값입니다. 셸에서 실행한 직전 명령의 종료 코드는 $?로 확인할 수 있습니다.

$ ls /not-exist
ls: /not-exist: No such file or directory
$ echo $?
1 # ← 명령 수행 실패를 의미하는 코드 1 반환

POSIX 규격에서 통용되는 주요 exit code의 의미는 다음과 같습니다.

0은 성공을 의미하지만, 0이 아닌 나머지 코드의 구체적인 의미는 애플리케이션 단위에서 정의하여 사용할 수 있습니다. 즉, "exit code 2는 오류가 아니라 데이터 미존재(스킵)를 의미한다"와 같은 규칙 설정이 가능합니다. 앞서 설계한 크롤러가 사용한 방식이 바로 이것입니다.

3. 오퍼레이터별 기능 비교

두 환경의 결과 차이는 Airflow 오퍼레이터의 기능 공백에서 발생했습니다. Airflow 소스 코드를 확인한 결과, 컨테이너 실행 오퍼레이터 간 옵션 지원 여부가 달랐습니다.

skip_on_exit_code는 지정한 exit code가 반환되었을 때 AirflowSkipException을 발생시켜 태스크를 스킵 상태로 처리하는 파라미터입니다. 타 오퍼레이터와의 기능 일관성을 맞추기 위해, EcsRunTaskOperator에도 해당 기능을 직접 추가하기로 결정했습니다.

4. 기존 PR 보완 및 코드 구현

동일한 기능 구현을 시도했던 기존 작업(PR #47256)이 존재했으나, 머지되지 못하고 중단된 상태임을 확인했습니다. 그래서 이 PR의 구현 방향을 참고하여, 누락된 부분까지 보완해 직접 구현한 뒤 새 PR(apache/airflow#63274)로 제출했습니다.

수정 작업은 providers/amazon/.../operators/ecs.py 파일에서 진행되었습니다.

# 1) 파라미터 정의 — 단일 정수(int) 또는 컨테이너 객체 허용
skip_on_exit_code: int | Container[int] | None = None,

# 2) 데이터 정규화 — 단일 값 리스트화 및 None 예외 처리
self.skip_on_exit_code = (
    skip_on_exit_code
    if isinstance(skip_on_exit_code, Container)
    else [skip_on_exit_code]
    if skip_on_exit_code is not None
    else []
)

# 3) 조건 판정 — 지정된 exit code 일치 여부에 따라 발생 예외를 분기
exit_code = container.get("exitCode", 1)
if exit_code in self.skip_on_exit_code:
    exception_cls: type[AirflowException] = AirflowSkipException
else:
    exception_cls = AirflowException

하드코딩되어 있던 AirflowException 발생 로직을, skip_on_exit_code 포함 여부에 따라 AirflowSkipException으로 분기되도록 수정한 것이 핵심입니다.

기능 검증을 위해 아래 항목에 대한 유닛 테스트(Unit Test)도 추가했습니다.

• 정의된 exit code 반환 시 AirflowSkipException 발생 여부
• 로그 출력과 스킵 처리의 정상 연동 여부
• 정의되지 않은 exit code 반환 시 기존대로 AirflowException 발생 여부

5. 마무리

• DockerOperator와 KubernetesPodOperator에서 제공하던 skip_on_exit_code 옵션을 EcsRunTaskOperator에도 동일하게 구현했습니다.
• 이를 통해 AWS ECS 환경에서도 특정 exit code를 기준으로 태스크를 실패가 아닌 스킵(Skip) 상태로 제어할 수 있게 되었습니다.
• 오픈소스 기여 시, 중단된 기존 PR을 분석하고 보완하는 방식으로 개발 리소스를 단축할 수 있습니다. 새로운 문제일수록 누군가 이미 시도했을 가능성이 높으므로, 구현에 앞서 관련 이슈와 PR을 먼저 검색하는 것이 효율적입니다.

전체 구현과 리뷰 과정은 머지된 PR에서 확인할 수 있습니다 → apache/airflow#63274 — Add skiponexit_code support to EcsRunTaskOperator