실패 처리: retry budget, alert fatigue, partial failure, compensation

파이프라인이 실패하는 건 예외가 아니라 정상이다

프로덕션 데이터 파이프라인에서 task 실패는 예외 상황이 아니다. 네트워크 순단, 외부 API 일시 오류, 메모리 부족 — 이런 일은 규모가 커질수록 더 자주 일어난다. 문제는 실패 자체가 아니라 실패를 어떻게 처리하느냐다.

잘못된 실패 처리는 두 가지 방향으로 망가진다. 첫째, 재시도를 너무 공격적으로 설정해 하나의 실패가 수십 개의 알림을 만들고 팀이 알림에 무감각해진다. 둘째, 실패 전파가 잘못 설계되어 일부 task 실패가 DagRun 성공으로 기록되거나 반대로 정상 task도 모두 막아버린다.

이 편에서는 네 가지 주제를 다룬다.

1. Retry 설정: 재시도 파라미터와 지수 백오프
2. Retry budget: 전체 재시도 횟수를 의식적으로 제한하는 패턴
3. Alert fatigue: 재시도가 만들어내는 알림 피로도와 해결 전략
4. Partial failure / Compensation: 부분 실패 전파와 보상 트랜잭션 패턴

1. Retry 설정: 기본 파라미터

Airflow의 재시도는 BaseOperator에 정의된 파라미터로 제어한다.

from datetime import timedelta

default_args = {
    "retries": 3,                          # 최대 재시도 횟수 (기본값: 0)
    "retry_delay": timedelta(minutes=5),   # 재시도 간 대기 시간
    "retry_exponential_backoff": True,     # 지수 백오프 활성화
    "max_retry_delay": timedelta(hours=1), # 재시도 대기 최대 상한
    "execution_timeout": timedelta(hours=2), # task 단건 실행 타임아웃
    "email_on_retry": False,               # 재시도 시 이메일 발송 비활성화 ← 중요
    "email_on_failure": True,             # 최종 실패 시에만 이메일
}

1.1 지수 백오프 계산 방식

재시도 간격이 지수적으로 증가하는 로직은 Airflow 내부 코드의 실제 공식을 따른다.

대기 시간 = min(retry_delay × 2^(try_number-1), max_retry_delay)

예: retry_delay=5분, max_retry_delay=1시간
1회차 재시도: 5분 × 2^0 = 5분
2회차 재시도: 5분 × 2^1 = 10분
3회차 재시도: 5분 × 2^2 = 20분
4회차 재시도: 5분 × 2^3 = 40분 → max_retry_delay 1시간 이내이므로 40분

Airflow 3.2+ 변화: retry_exponential_backoff가 boolean 대신 float 배수를 받는다.

# Airflow 3.2+
"retry_exponential_backoff": 2.0,   # True의 기존 동작과 동일
"retry_exponential_backoff": 3.0,   # 더 공격적인 백오프

1.2 파라미터 우선순위

airflow.cfg [core] default_task_retries
    ↓ 재정의
DAG의 default_args
    ↓ 재정의
개별 task 파라미터

환경변수로 클러스터 전체 상한을 설정할 수 있다.

AIRFLOW__CORE__MAX_TASK_RETRY_DELAY=86400   # 어떤 task도 24시간 이상 대기 불가

1.3 task 중요도별 차등 설정

모든 task에 같은 재시도 정책을 적용하는 것은 비효율적이다. 중요도와 실패 특성에 따라 나눈다.

# 핵심 task: 재시도 많고 지수 백오프
CRITICAL_ARGS = {
    "retries": 5,
    "retry_delay": timedelta(minutes=2),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
}

# 외부 API 호출: 빠른 재시도 + 짧은 타임아웃
API_CALL_ARGS = {
    "retries": 3,
    "retry_delay": timedelta(seconds=30),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=10),
    "execution_timeout": timedelta(minutes=5),
}

# 낮은 우선순위 task: 재시도 최소화
BEST_EFFORT_ARGS = {
    "retries": 1,
    "retry_delay": timedelta(minutes=10),
}

2. Retry budget: 과도한 재시도 제한

2.1 문제: 재시도가 폭증하는 상황

100개 task × 5회 재시도 = 최대 500번의 재시도가 하나의 DagRun에서 발생할 수 있다. 인프라 장애처럼 근본 원인이 해결되지 않은 상태에서 모든 task가 재시도를 반복하면 외부 시스템에 부하를 가중시키고 클러스터 리소스를 낭비한다.

Retry budget은 Airflow에 내장된 기능이 아니다. 파이프라인이 소비할 수 있는 총 재시도 횟수를 사전에 정의하고, 이를 초과하면 추가 재시도를 막는 운영 패턴이다.

2.2 구현: on_retry_callback 기반 budget 추적

from airflow.models import XCom

PIPELINE_RETRY_BUDGET = 10  # 파이프라인 전체 최대 재시도

def budget_aware_retry_callback(context):
    ti = context["task_instance"]
    dag_run = context["dag_run"]

    current = XCom.get_one(
        dag_id=dag_run.dag_id,
        run_id=dag_run.run_id,
        task_id="__retry_budget__",
        key="total_retries",
    ) or 0
    current += 1

    if current >= PIPELINE_RETRY_BUDGET:
        # 예산 소진 → 즉각 에스컬레이션, 더 이상 재시도 안 함
        ti.max_tries = ti.try_number  # 강제 소진
        ti.save()
        send_critical_alert(f"Retry budget exhausted for {dag_run.dag_id}")
    else:
        XCom.set(
            dag_id=dag_run.dag_id,
            run_id=dag_run.run_id,
            task_id="__retry_budget__",
            key="total_retries",
            value=current,
        )

실용적인 환경에서는 XCom 대신 Redis나 외부 DB로 카운터를 관리하는 것이 안전하다.

3. Alert fatigue: 알림이 많을수록 놓친다

3.1 재시도가 만들어내는 알림 폭탄

핵심 원칙: 사람이 행동해야 할 때만 알림이 와야 한다. 재시도는 시스템이 스스로 회복하는 과정이므로 운영자의 즉각 대응을 요구하지 않는다. 최종 실패만이 행동을 요구한다.

3.2 콜백으로 알림 채널 분리

Airflow에는 task 생명주기마다 실행되는 콜백 함수가 있다.

def on_retry(context):
    """재시도 시: 낮은 우선순위 채널에 조용히 기록"""
    ti = context["task_instance"]
    post_slack(
        channel="#airflow-debug",       # 온콜 팀 미포함
        text=f"재시도 중: {ti.dag_id}.{ti.task_id} "
             f"({ti.try_number}/{ti.max_tries})",
    )

def on_failure(context):
    """최종 실패 시: 고우선순위 채널 + 온콜"""
    ti = context["task_instance"]
    post_slack(
        channel="#data-oncall",
        text=f"[장애] {ti.dag_id}.{ti.task_id} 재시도 소진",
        mention="@oncall",
    )
    send_pagerduty(context)

default_args = {
    "on_retry_callback": on_retry,
    "on_failure_callback": on_failure,
    "email_on_retry": False,     # 재시도 이메일 비활성화
    "email_on_failure": True,    # 최종 실패만 이메일
}

Airflow 2.6+부터 콜백 리스트가 지원된다.

"on_failure_callback": [send_slack, send_pagerduty, log_incident],

3.3 콜백 context 주요 키

def my_callback(context: dict):
    ti = context["task_instance"]
    exception = context.get("exception")       # 발생한 예외 객체
    dag_run = context["dag_run"]               # DagRun 정보
    # Airflow 3.0+: dag_run은 Pydantic 모델, SQLAlchemy ORM 메서드 사용 불가
    
    is_final = ti.try_number >= ti.max_tries   # 최종 실패 여부 판별
    ds = context["ds"]                         # YYYY-MM-DD
    data_interval_start = context["data_interval_start"]

Airflow 3.0 주의: context["dag_run"]이 SQLAlchemy ORM 객체에서 Pydantic 모델로 변경됐다. DB 직접 접근 코드는 Task Execution API로 마이그레이션해야 한다.

3.4 SLA miss → Deadline Alerts (Airflow 3.0+)

Airflow 2.x의 SLA(sla, sla_miss_callback)가 Airflow 3.0에서 제거됐다. Airflow 3.1부터 Deadline Alerts가 공식 대체제다.

# Airflow 2.x 방식 (deprecated)
def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    ...

with DAG(..., sla_miss_callback=sla_miss_callback) as dag:
    task = PythonOperator(..., sla=timedelta(hours=2))

# Airflow 3.1+ 방식
from airflow.sdk import DAG, DeadlineAlert, DeadlineReference

async def deadline_callback(**kwargs):
    context = kwargs.get("context", {})
    dag_run = context.get("dag_run")
    print(f"DEADLINE EXCEEDED: {dag_run.dag_id}")

with DAG(
    dag_id="critical_pipeline",
    schedule="0 2 * * *",
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
        interval=timedelta(hours=1),   # logical_date 기준 1시간 이내 완료 필요
        callback=deadline_callback,    # async 함수 필수
    ),
) as dag:
    ...

SLA와 달리 Deadline은 만료 즉시(스케줄러 heartbeat 이내) 콜백이 호출된다.

4. Partial failure: 부분 실패 전파와 DagRun 상태

4.1 DagRun 상태는 리프 노드가 결정한다

Airflow의 DagRun 최종 상태는 리프 노드(다운스트림 없는 마지막 task) 의 상태로 결정된다. 중간 task가 실패해도 리프 노드가 성공이면 DagRun은 success가 된다. 이것이 Watcher 패턴이 필요한 이유다.

4.2 Watcher 패턴 구현

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule

def fail_if_called(**kwargs):
    raise Exception("Pipeline has failed tasks — see individual task logs")

with DAG(dag_id="watcher_demo", schedule=None) as dag:
    t1 = PythonOperator(task_id="extract", python_callable=lambda: None)
    t2 = PythonOperator(task_id="transform", python_callable=lambda: 1/0)
    t3 = PythonOperator(task_id="load", python_callable=lambda: None)

    cleanup = EmptyOperator(
        task_id="cleanup",
        trigger_rule=TriggerRule.ALL_DONE,  # 성공/실패 무관 항상 실행
    )

    watcher = PythonOperator(
        task_id="watcher",
        python_callable=fail_if_called,
        trigger_rule=TriggerRule.ONE_FAILED,  # 하나라도 실패 시 실행
    )

    [t1, t2, t3] >> cleanup
    [t1, t2, t3, cleanup] >> watcher

동작: t2 실패 → watcher 실행 → watcher도 실패(의도적) → DagRun = failed. t1, t2, t3 모두 성공 → watcher는 SKIPPED → DagRun = success.

4.3 depends_on_past와 블로킹 함정

with DAG(
    dag_id="sequential_pipeline",
    depends_on_past=True,   # 이전 실행의 같은 task가 성공해야 현재 실행 가능
    max_active_runs=1,
    dagrun_timeout=timedelta(hours=4),
) as dag:
    ...

depends_on_past=True가 설정된 DAG에서 이전 DagRun이 실패하면, 다음 DagRun의 동일 task가 영구 블로킹된다. 실패한 task를 수동으로 mark success 하거나 clear해야 이후 실행이 풀린다.

4.4 타임아웃 계층

5. Compensation 패턴: 실패 시 사이드 이펙트 정리

5.1 Setup/Teardown 프레임워크 (Airflow 2.7+)

리소스를 프로비저닝하는 task가 있을 때, 중간에 실패해도 반드시 리소스를 해제해야 한다. Setup/Teardown 프레임워크가 이 패턴을 공식 지원한다.

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

def create_cluster():
    return {"cluster_id": "spark-001"}

def delete_cluster():
    pass  # 실제 종료 로직

def process_data():
    pass

with DAG(dag_id="setup_teardown_demo", schedule=None) as dag:

    setup = PythonOperator(
        task_id="create_cluster",
        python_callable=create_cluster,
    ).as_setup()  # setup 태스크로 마킹

    teardown = PythonOperator(
        task_id="delete_cluster",
        python_callable=delete_cluster,
    ).as_teardown(
        setups=[setup],              # 연결된 setup 지정
        on_failure_fail_dagrun=True, # teardown 실패도 DagRun 실패로 마킹
    )

    work = PythonOperator(task_id="process_data", python_callable=process_data)

    # context manager 문법
    with teardown(setup):
        setup >> work

5.2 on_failure_callback을 이용한 롤백

cleanup이 별도 task로 필요 없는 경우, 실패 콜백에서 직접 사이드 이펙트를 정리한다.

import boto3

def rollback_s3(context):
    """업로드된 부분 파일 정리"""
    ti = context["task_instance"]
    uploaded_keys = ti.xcom_pull(task_ids="upload_to_s3", key="uploaded_keys")

    if uploaded_keys:
        s3 = boto3.client("s3")
        for key in uploaded_keys:
            s3.delete_object(Bucket="my-bucket", Key=key)

upload = PythonOperator(
    task_id="upload_to_s3",
    python_callable=upload_data,
    on_failure_callback=rollback_s3,
)

5.3 보상 task 패턴 (Saga 유사)

분산 트랜잭션처럼, 이미 성공한 step의 사이드 이펙트를 보상 task에서 역방향으로 취소한다.

from airflow.utils.trigger_rule import TriggerRule

def compensate(**kwargs):
    """부분 성공한 step의 사이드 이펙트 취소"""
    ti = kwargs["ti"]
    step1_ids = ti.xcom_pull(task_ids="step1", key="db_record_ids")
    if step1_ids:
        db.delete_records(step1_ids)

with DAG(dag_id="saga_compensation_demo") as dag:
    step1 = PythonOperator(task_id="step1", python_callable=insert_to_db)
    step2 = PythonOperator(task_id="step2", python_callable=send_api)
    step3 = PythonOperator(task_id="step3", python_callable=update_cache)

    compensate_task = PythonOperator(
        task_id="compensate",
        python_callable=compensate,
        trigger_rule=TriggerRule.ONE_FAILED,  # 하나라도 실패 시 보상 실행
    )

    [step1, step2, step3] >> compensate_task

6. 프로덕션 권장 설정 요약

from datetime import timedelta
from airflow.sdk import DAG, DeadlineAlert, DeadlineReference

async def deadline_callback(**kwargs):
    ...

PRODUCTION_ARGS = {
    # 재시도
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(hours=1),

    # 알림 피로도 방지
    "email_on_retry": False,
    "email_on_failure": True,
    "on_retry_callback": log_retry_quietly,
    "on_failure_callback": escalate_to_oncall,

    # 타임아웃
    "execution_timeout": timedelta(hours=2),
}

with DAG(
    dag_id="production_dag",
    default_args=PRODUCTION_ARGS,
    dagrun_timeout=timedelta(hours=6),
    deadline=DeadlineAlert(           # Airflow 3.1+ SLA 대체
        reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
        interval=timedelta(hours=4),
        callback=deadline_callback,
    ),
    catchup=False,
    schedule="@daily",
) as dag:
    ...

6.1 운영 체크리스트

마무리: 실패를 설계 범위 안에 두기

재시도, 알림, 부분 실패 전파, 보상 — 모두 "실패가 발생했을 때 시스템이 어떻게 반응해야 하는가"라는 하나의 질문에서 나온다. 이 질문에 답을 준비하지 않으면, 실패가 발생한 순간에 운영자가 대신 답을 만들어야 하고, 그 과정에서 데이터가 손상되거나 서비스가 지연된다.

핵심은 간단하다. 자동으로 회복 가능한 실패는 조용하게, 사람의 판단이 필요한 실패만 크게 알린다. 그리고 실패한 task가 만든 사이드 이펙트는 반드시 시스템이 정리한다.

References

• Apache Airflow Documentation, "Callbacks": https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/callbacks.html
• Apache Airflow Documentation, "Deadline Alerts": https://airflow.apache.org/docs/apache-airflow/stable/howto/deadline-alerts.html
• Apache Airflow Documentation, "Migrating from SLA to Deadline Alerts": https://airflow.apache.org/docs/apache-airflow/stable/howto/sla-to-deadlines.html
• Apache Airflow Documentation, "Setup and Teardown": https://airflow.apache.org/docs/apache-airflow/stable/howto/setup-and-teardown.html
• Apache Airflow Documentation, "Best Practices": https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html
• Apache Airflow Documentation, "DAG Runs": https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html
• Apache Airflow Blog, "Apache Airflow 3 is Generally Available": https://airflow.apache.org/blog/airflow-three-point-oh-is-here/
• Apache Airflow Blog, "Apache Airflow 3.1.0: Human-Centered Workflows": https://airflow.apache.org/blog/airflow-3.1.0/
• AIP-86 Deadline Alerts: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=323488182
• Astronomer, "Manage Apache Airflow DAG Notifications": https://www.astronomer.io/docs/learn/error-notifications-in-airflow
• Astronomer, "Introducing Apache Airflow® 3.2": https://www.astronomer.io/blog/apache-airflow-3-2-release/
• GitHub Issue #47971, "retry_exponential_backoff OverflowError": https://github.com/apache/airflow/issues/47971