날씨 캐시 DAG가 하루 정도는 잘 돌다가 어느 시점부터 429 Too Many Requests로 계속 실패했다.
처음에는 일시적인 외부 API 장애로 봤는데, 로그와 호출 수를 같이 계산해보니 문제는 훨씬 단순했다.
대기질 API 일일 한도를 넘긴 상태에서, 재시도 없이 전체 태스크를 바로 실패시키고 있었다.
Airflow 날씨 갱신이 왜 계속 실패했나
실패 로그는 이렇게 찍혔다.
requests.exceptions.HTTPError: 429 Client Error: Too Many Requests for url: [도메인]
앞부분 로그를 같이 보면 더 명확하다.
에어코리아 대기질 캐시 갱신 완료: 강원 stations=40
에어코리아 대기질 캐시 갱신 완료: 경남 stations=50
에어코리아 대기질 캐시 갱신 완료: 경북 stations=53
에어코리아 대기질 캐시 갱신 완료: 광주 stations=13
Task failed with exception
requests.exceptions.HTTPError: 429 Client Error: Too Many Requests for url: [도메인]
여기서 확인한 사실은 세 가지다.
날씨 API가 아니라 대기질 API 호출에서 실패했다.
몇 개 시도는 이미 성공한 뒤 대구 요청에서 막혔다.
response.raise_for_status() 때문에 한 번의 429가 전체 DAG 실패로 번졌다.
지역마다 호출하는 줄 알았는데 실제 호출 수는 몇 번이었나
처음에는 도시별로 다 때리는 줄 알았는데 코드를 끝까지 읽어보니 그렇지는 않았다.
현재 구조는 이렇다.
도시 수: 37
중복 제거한 시도 수: 16
DAG 주기: 30분마다
대기질 API 호출: 실행당 16회
즉 하루 호출 수는 아래처럼 계산된다.
16회 x 48번 = 768회/일
문제는 개발계정 기준 일일 트래픽이 500건일 때 이 수치를 초과한다는 점이다.
768회/일이면 자정부터 돌기 시작했을 때 오후쯤 한도에 걸리는 패턴이 딱 나온다.
왜 하루 정도 잘되다가 그 뒤로는 계속 실패했나
이 패턴은 순간적인 트래픽 급증보다 일일 quota 초과로 보는 쪽이 더 맞았다.
실제 흐름은 대체로 이렇다.
자정 이후 스케줄대로 정상 호출이 누적된다.
누적 호출 수가 일일 한도에 도달한다.
특정 시도 요청부터 429가 떨어진다.
이후 같은 날짜 안에서는 남은 스케줄도 계속 429가 난다.
기존 코드는 이를 흡수하지 못해 매번 태스크 전체가 실패한다.
즉 429를 보낸 주체는 외부 대기질 API이고, 계속 실패한 이유는 우리 쪽 DAG가 한도 초과 상태를 고려하지 않았기 때문이다.
처음에 헷갈렸던 지점
중간에 가장 많이 헷갈렸던 부분은 이거였다.
정말 도시 37개를 각각 호출하는가
아니면 시도 16개만 호출하는가
429가 순간 제한인가
아니면 하루 누적 한도인가
코드를 읽고 정리한 결과는 아래였다.
항목 | 실제 동작 |
|---|
지역 단위 | 도시 37개 |
대기질 조회 단위 | 시도 16개 |
실행당 대기질 호출 | 16회 |
스케줄 | */30 * * * *
|
하루 호출 수 | 768회 |
실패 형태 | 429 발생 시 전체 태스크 실패
|
도시 37개를 직접 치는 구조는 아니었지만, 시도 16개를 30분마다 계속 치는 것만으로도 충분히 한도 초과가 날 수 있었다.
어떻게 수정했나
수정 방향은 새 구조를 만드는 게 아니라 기존 흐름을 유지하면서 최소 단위만 바꾸는 쪽으로 잡았다.
바꾼 내용은 두 가지다.
1. 429 재시도와 요청 간격 추가
AIRKOREA_REQUEST_INTERVAL_SECONDS = 1.2
AIRKOREA_MAX_RETRIES = 3
AIRKOREA_RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
AIRKOREA_RETRY_BACKOFF_SECONDS = 3
def fetch_airkorea_json(session, endpoint, params):
last_requested_at = getattr(session, '_airkorea_last_requested_at', None)
if last_requested_at is not None:
elapsed = time.monotonic() - last_requested_at
if elapsed < AIRKOREA_REQUEST_INTERVAL_SECONDS:
time.sleep(AIRKOREA_REQUEST_INTERVAL_SECONDS - elapsed)
for attempt in range(AIRKOREA_MAX_RETRIES + 1):
response = session.get(
f'{AIRKOREA_API_BASE}/{endpoint}',
params=params,
timeout=AIRKOREA_TIMEOUT_SECONDS,
)
session._airkorea_last_requested_at = time.monotonic()
if response.status_code not in AIRKOREA_RETRY_STATUS_CODES:
response.raise_for_status()
return response.json()
if attempt >= AIRKOREA_MAX_RETRIES:
response.raise_for_status()
time.sleep(AIRKOREA_RETRY_BACKOFF_SECONDS * (attempt + 1))</code></pre><h3>2. 한 시도 실패가 전체 실패가 되지 않게 변경</h3><pre><code class="language-python">for sido_name in sorted({str(region['sido_name']) for region in WEATHER_REGIONS}):
try:
stations = fetch_airkorea_stations(session, airkorea_service_key, sido_name)
except requests.RequestException:
logger.exception('에어코리아 대기질 캐시 갱신 실패: %s', sido_name)
stations = []</code></pre><p>이렇게 바꾸면 특정 시도에서 실패해도 전체 캐시 작업은 계속 진행된다.</p><h3>3. DAG 실행 주기를 1시간으로 변경</h3><pre><code class="language-python">with DAG(
'weather_cache_update',
default_args=default_args,
description='주요 도시 현재 날씨와 미세먼지를 1시간마다 MongoDB에 캐시',
schedule_interval='0 * * * *',
catchup=False,
) as dag:
이 변경의 핵심은 호출량 계산이다.
기존: 16회 x 48번 = 768회/일
변경: 16회 x 24번 = 384회/일
384회/일이면 개발계정 500건/일 기준 아래로 내려간다.
1시간 간격으로 바꾸면 바로 해결되나
완전히 즉시 해결된다고 보긴 어렵다.
이유는 아래와 같다.
이미 그날 quota를 넘긴 뒤라면 리셋 전까지는 계속 429가 날 수 있다.
같은 [API_KEY]를 다른 작업에서도 쓰면 합산된다.
재시도까지 포함하면 실제 호출 수는 계산치보다 조금 늘 수 있다.
그래도 30분에서 1시간으로 낮추는 건 의미가 크다.
삽질하면서 알게 된 것
Q1. 429면 무조건 순간 트래픽 문제인가
아니다. 이번 경우처럼 하루 누적 호출량이 쌓여서 막히는 경우도 많다.
Q2. 도시 37개를 다 호출해서 터진 건가
아니다. 실제로는 시도 16개만 호출하고 있었다.
Q3. 한 시도만 실패했는데 왜 전체 DAG가 죽었나
response.raise_for_status() 예외를 바로 올렸고, 바깥에서 흡수하지 않았기 때문이다.
Q4. 30분 주기에서 1시간 주기로 바꾸면 호출 수는 얼마나 줄어드나
768회/일에서 384회/일로 줄어든다.
Q5. 그래도 429가 날 수 있나
그렇다. 같은 키를 다른 곳에서도 쓰면 합산 제한에 걸릴 수 있다.
Q6. 가장 먼저 확인할 건 무엇인가
에러 로그만 보지 말고 실행당 호출 수 x 하루 실행 횟수를 바로 계산해보는 것이다. 이번 건도 원인은 로그보다 계산이 더 빨리 설명해줬다.