RTB: 수집 파이프라인 구조 개선안

RTB: 수집 파이프라인 구조 개선

배경

얼마 전까지 신규 기능 개발과 리팩토링을 주로 하다가, 운영성 업무를 맡게 되었다.
참고로 나는 이런 종류의 일을 RTB(Run the Business)라고 부른다(새로운 것을 만드는 게 아니라, 이미 돌아가고 있는 것을 유지하고 확장하는 일)

 

보통 운영 업무는 "어쩔 수 없는 것"으로 받아들이고 단순히 유지보수만 하게 되기 쉽다.
그게 답답해서 개선 포인트를 좀 찾아보았다. 대상은 50개가 넘는 외부 API에서 데이터를 수집하는 파이프라인이다.

이 글은 그 과정에서 발견한 구조적 문제들과, 개선 방향을 설계한 기록이다. 실제 구현은 다음 주 중에 팀에 제안하고 진행할 예정이다.

---

매일 아침 하는 일

새벽 배치가 돌고 나면 Slack에 리포트가 온다. "확인 필요"가 뜨면 admin에 들어가서 에러 상태인 데이터 연결을 하나씩 확인한다.

확인하는 건 크게 네 가지다:

• 인증 문제 — 고객이 토큰을 만료시켰거나, 비밀번호를 바꿨거나
• API 측 문제 — 매체 서버가 죽었거나, rate limit에 걸렸거나
• 코드 문제 — 수집기 버그
• 데이터 문제 — 예상 못한 형식의 응답

문제는, 이 판단을 사람이 매일 반복하고 있다는 거다.

데이터 연결의 상태값은 complete과 error 두 종류뿐이다. 인증 만료든 코드 버그든 API 장애든 전부 같은 error. 원인을 알려면 수집 태스크의 텍스트 로그를 직접 열어서 읽어야 한다.

더 나쁜 건, 고객이 계정 정보를 잘못 입력하고 방치한 경우다. 이런 연결은 한번 에러에 빠지면 고객이 고치기 전까지 매일 같은 에러가 반복해서 올라온다. 어제 확인한 에러를 오늘 또 보고, 내일 또 본다.

---

코드베이스를 뜯어보니

에러 분류 체계는 이미 잘 만들어져 있었다:

CollectorAuthenticationError   — 인증 정보 오류
CollectorWrongPasswordError    — 비밀번호 오류
CollectorLoginError            — 로그인 실패
CollectorTimeoutError          — 타임아웃
CollectorUnhandledError        — 알 수 없는 오류
PreprocessDataError            — 전처리 오류
...

9개의 에러 클래스가 명확하게 나뉘어 있다. 수집기에서 에러가 발생하면 이 중 하나로 raise된다.

그런데 이 정보가 저장되지 않는다. 에러를 catch하는 시점에서 로그에 텍스트로 남기고, 데이터 연결 상태는 그냥 error로 바뀐다. 에러 클래스가 9개로 나뉘어 있는데, 최종적으로 Connection에 도달하는 순간 전부 하나의 문자열 "error"가 된다.

# 현재 코드 — 에러 유형과 무관하게 동일한 처리
except (CollectorAuthenticationError, CollectorWrongPasswordError) as e:
    await self.change_connection_status(Connection.CONN_ERROR)

except Exception:
    await self.change_connection_status(Connection.CONN_ERROR)

Slack 알림도 마찬가지다. 에러 상태인 데이터소스 목록만 나열할 뿐, 유형은 없다:

오류 발생 데이터소스: cafe24, shopify, coupang_ads
https://admin/connections/connection/?status__exact=error

이 알림을 받고 할 수 있는 건 — 링크를 클릭해서 admin에 들어가는 것뿐이다.

---

HTTP 클라이언트 쪽도 문제였다

에러 모니터링과 별개로, 수집기 코드 전체를 분석해봤더니 HTTP 클라이언트 레벨에서도 문제가 있었다.

• 20개 async 수집기에서 매 요청마다 새 aiohttp.ClientSession을 생성하고 있었다 (약 50회). TCP connection pooling이 사실상 무효화된 상태
• HTTP 429(rate limit) 감지 + 재시도 로직이 전체 수집기 중 1곳에만 존재
• async 함수 내에서 time.sleep()을 사용하는 곳이 있었다 — event loop 전체가 블로킹됨
• 수집기마다 timeout 설정이 제각각 (30초 ~ 미설정(기본 300초))
• retry 라이브러리가 설치되어 있지만 코드에서 사용하는 곳이 없음

50개 수집기가 각자 자기만의 방식으로 HTTP 요청을 보내고 있었다. 공통 패턴이 없으니, 새 수집기를 만들 때마다 기존 것 하나를 골라서 복사한 뒤 수정하는 식이었다. 어떤 건 세션을 재사용하고, 어떤 건 안 하고, 어떤 건 429를 처리하고, 대부분은 안 했다.

---

개선 방향

에러 모니터링: "매일 전체를 보는 것"에서 "신규만 보는 것"으로

핵심은 세 가지다:

1. 에러 분류를 저장한다

데이터 연결에 에러 카테고리 필드를 추가하고, 수집 태스크에서 에러를 catch할 때 분류값을 함께 저장한다. 이미 있는 9개 에러 클래스를 5개 카테고리로 매핑하면 된다.

auth     — 인증 오류 (AuthenticationError, WrongPasswordError, LoginError)
timeout  — 타임아웃 (TimeoutError)
data     — 데이터 오류 (DataNotFoundError, PreprocessDataError, DownloadError)
execution — 실행 오류 (ExecutionError)
unknown  — 알 수 없는 오류 (UnhandledError, 기타)

이것만으로도 admin에서 로그를 열어보지 않고 에러 유형을 판단할 수 있다.

2. Slack 알림에 유형과 신규 여부를 표시한다

데이터 수집 오류 (신규 2건 / 기존 5건)

[신규]
- cafe24 | ClientA | 타임아웃
- shopify | ClientC | 인증 오류

[기존 — 미조치]
- coupang_ads | ClientB | 인증 오류 (3일째)

신규 에러만 빠르게 확인하고, 기존 에러는 건수만 파악하면 된다.

3. 확인 완료된 에러는 알림에서 뺀다

고객사 측 문제로 확인된 에러에 "확인 완료" 표시를 하면, 다음 날부터 알림에 올라오지 않는다. 수집이 성공하면 자동으로 리셋된다.

HTTP 클라이언트: 공통 레이어 도입

이쪽은 3단계로 접근하려 한다:

• Phase 1: 공통 HTTP client wrapper + 세션 재사용 패턴 표준화
• Phase 2: 데이터소스별 rate limit 설정 + asyncio.Semaphore 기반 동시 요청 제한 + 429 exponential backoff
• Phase 3 (선택): Redis 기반 분산 rate limiter — 여러 worker가 동시에 수집할 때만 필요

Phase 1만으로도 TCP 연결 재사용과 에러 분류가 개선된다. 현재 1곳에만 있는 429 재시도 패턴을 공통 레이어로 올리면, 새 수집기를 만들 때 이런 것들을 매번 고민할 필요가 없어진다.

---

마무리

솔직히 운영 업무를 처음 맡았을 때는 어필이 안 되는 일이라고 생각했다.

신규 기능과 달리, 잘해도 티가 안 나고 못하면 바로 티가 나는 영역이니까.

하지만 3주간 매일 에러를 분류하면서 "이건 사람이 할 일이 아니다"라는 결론에 도달했고, 그게 이 글의 시작이 되었다.

 

수집기 하나를 새로 만드는 건 며칠이면 끝난다. 하지만 50개가 매일 돌아가면서 생기는 문제를 사람이 수동으로 분류하고 판단하는 구조는, 수집기가 60개, 70개가 되어도 그대로 사람에게 남는다.

 

결국 하려는 건 운영에서 반복되는 판단을 구조화하는 것이다. 대단한 기술이 필요한 건 아니었다. 매일 하는 일을 관찰하고, "이건 사람이 할 일이 아니다"라고 판단하는 것에서 시작했을 뿐이다. 빠르게 이런 운영성 업무들을 최적화하고 신규 기능 개발 및 확장에 집중하는 방향으로 바꾸고 싶다. 그로부터 훨씬 효율적으로 일하고 싶다