로컬은 스킵, ECS에서는 실패 — Airflow EcsRunTaskOperator에 skip_on_exit_code 추가하기

관리비 및 공과금 데이터를 크롤링하는 파이프라인을 구축했습니다. 로컬 개발 환경에서는 크롤러를 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